Compare commits
31 Commits
Author | SHA1 | Date | |
---|---|---|---|
![]() |
c9504f61fc | ||
![]() |
75c159baa8 | ||
![]() |
41ece2cf2d | ||
![]() |
5e6adfac06 | ||
![]() |
b163084a5f | ||
![]() |
ad7db2bb1e | ||
![]() |
b5dc2266a6 | ||
![]() |
ba233791e1 | ||
![]() |
0ce2ef14a1 | ||
![]() |
4db8b94cca | ||
![]() |
734e4cf8e6 | ||
![]() |
dcf30b1c54 | ||
![]() |
065053d859 | ||
![]() |
1935a663df | ||
![]() |
2c7eb87c85 | ||
![]() |
1674e682fe | ||
![]() |
7c47afd7d2 | ||
![]() |
3e0cc1e717 | ||
![]() |
6fa95eb497 | ||
![]() |
ba4a7e004b | ||
![]() |
4bd81d0933 | ||
![]() |
f690f3a425 | ||
![]() |
af6f459a23 | ||
![]() |
3ac81f3ae2 | ||
![]() |
4ace7c7d77 | ||
![]() |
a09874b40c | ||
![]() |
a5437f246b | ||
![]() |
f272557516 | ||
![]() |
71eba353d2 | ||
![]() |
557eee826f | ||
![]() |
b71df1f814 |
13
Makefile
13
Makefile
@@ -20,12 +20,16 @@ clean:
|
||||
rm -f ./codecov
|
||||
rm -rf ./agent-*
|
||||
rm -rf ./covdir
|
||||
rm -f ./*.coverprofile
|
||||
rm -f ./*.log
|
||||
rm -f ./bin/Dockerfile-release
|
||||
rm -rf ./bin/*.etcd
|
||||
rm -rf ./default.etcd
|
||||
rm -rf ./tests/e2e/default.etcd
|
||||
rm -rf ./gopath
|
||||
rm -rf ./gopath.proto
|
||||
rm -rf ./release
|
||||
rm -f ./snapshot/localhost:*
|
||||
rm -f ./integration/127.0.0.1:* ./integration/localhost:*
|
||||
rm -f ./clientv3/integration/127.0.0.1:* ./clientv3/integration/localhost:*
|
||||
rm -f ./clientv3/ordering/127.0.0.1:* ./clientv3/ordering/localhost:*
|
||||
@@ -46,7 +50,8 @@ docker-remove:
|
||||
|
||||
|
||||
|
||||
GO_VERSION ?= 1.10.1
|
||||
# GO_VERSION ?= 1.10.3
|
||||
GO_VERSION ?= 1.8.7
|
||||
ETCD_VERSION ?= $(shell git rev-parse --short HEAD || echo "GitNotFound")
|
||||
|
||||
TEST_SUFFIX = $(shell date +%s | base64 | head -c 15)
|
||||
@@ -61,16 +66,16 @@ endif
|
||||
|
||||
# Example:
|
||||
# GO_VERSION=1.8.7 make build-docker-test
|
||||
# GO_VERSION=1.9.5 make build-docker-test
|
||||
# GO_VERSION=1.9.7 make build-docker-test
|
||||
# make build-docker-test
|
||||
#
|
||||
# gcloud docker -- login -u _json_key -p "$(cat /etc/gcp-key-etcd-development.json)" https://gcr.io
|
||||
# GO_VERSION=1.8.7 make push-docker-test
|
||||
# GO_VERSION=1.9.5 make push-docker-test
|
||||
# GO_VERSION=1.9.7 make push-docker-test
|
||||
# make push-docker-test
|
||||
#
|
||||
# gsutil -m acl ch -u allUsers:R -r gs://artifacts.etcd-development.appspot.com
|
||||
# GO_VERSION=1.9.5 make pull-docker-test
|
||||
# GO_VERSION=1.9.7 make pull-docker-test
|
||||
# make pull-docker-test
|
||||
|
||||
build-docker-test:
|
||||
|
@@ -118,6 +118,11 @@ func (t *tokenSimple) genTokenPrefix() (string, error) {
|
||||
|
||||
func (t *tokenSimple) assignSimpleTokenToUser(username, token string) {
|
||||
t.simpleTokensMu.Lock()
|
||||
defer t.simpleTokensMu.Unlock()
|
||||
if t.simpleTokenKeeper == nil {
|
||||
return
|
||||
}
|
||||
|
||||
_, ok := t.simpleTokens[token]
|
||||
if ok {
|
||||
plog.Panicf("token %s is alredy used", token)
|
||||
@@ -125,7 +130,6 @@ func (t *tokenSimple) assignSimpleTokenToUser(username, token string) {
|
||||
|
||||
t.simpleTokens[token] = username
|
||||
t.simpleTokenKeeper.addSimpleToken(token)
|
||||
t.simpleTokensMu.Unlock()
|
||||
}
|
||||
|
||||
func (t *tokenSimple) invalidateUser(username string) {
|
||||
|
67
auth/simple_token_test.go
Normal file
67
auth/simple_token_test.go
Normal file
@@ -0,0 +1,67 @@
|
||||
// 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 auth
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestSimpleTokenDisabled ensures that TokenProviderSimple behaves correctly when
|
||||
// disabled.
|
||||
func TestSimpleTokenDisabled(t *testing.T) {
|
||||
initialState := newTokenProviderSimple(dummyIndexWaiter)
|
||||
|
||||
explicitlyDisabled := newTokenProviderSimple(dummyIndexWaiter)
|
||||
explicitlyDisabled.enable()
|
||||
explicitlyDisabled.disable()
|
||||
|
||||
for _, tp := range []*tokenSimple{initialState, explicitlyDisabled} {
|
||||
ctx := context.WithValue(context.WithValue(context.TODO(), "index", uint64(1)), "simpleToken", "dummy")
|
||||
token, err := tp.assign(ctx, "user1", 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
authInfo, ok := tp.info(ctx, token, 0)
|
||||
if ok {
|
||||
t.Errorf("expected (true, \"user1\") got (%t, %s)", ok, authInfo.Username)
|
||||
}
|
||||
|
||||
tp.invalidateUser("user1") // should be no-op
|
||||
}
|
||||
}
|
||||
|
||||
// TestSimpleTokenAssign ensures that TokenProviderSimple can correctly assign a
|
||||
// token, look it up with info, and invalidate it by user.
|
||||
func TestSimpleTokenAssign(t *testing.T) {
|
||||
tp := newTokenProviderSimple(dummyIndexWaiter)
|
||||
tp.enable()
|
||||
ctx := context.WithValue(context.WithValue(context.TODO(), "index", uint64(1)), "simpleToken", "dummy")
|
||||
token, err := tp.assign(ctx, "user1", 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
authInfo, ok := tp.info(ctx, token, 0)
|
||||
if !ok || authInfo.Username != "user1" {
|
||||
t.Errorf("expected (true, \"token2\") got (%t, %s)", ok, authInfo.Username)
|
||||
}
|
||||
|
||||
tp.invalidateUser("user1")
|
||||
|
||||
_, ok = tp.info(context.TODO(), token, 0)
|
||||
if ok {
|
||||
t.Errorf("expected ok == false after user is invalidated")
|
||||
}
|
||||
}
|
@@ -529,6 +529,20 @@ func isHaltErr(ctx context.Context, err error) bool {
|
||||
return ev.Code() != codes.Unavailable && ev.Code() != codes.Internal
|
||||
}
|
||||
|
||||
// isUnavailableErr returns true if the given error is an unavailable error
|
||||
func isUnavailableErr(ctx context.Context, err error) bool {
|
||||
if ctx != nil && ctx.Err() != nil {
|
||||
return false
|
||||
}
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
ev, _ := status.FromError(err)
|
||||
// Unavailable codes mean the system will be right back.
|
||||
// (e.g., can't connect, lost leader)
|
||||
return ev.Code() == codes.Unavailable
|
||||
}
|
||||
|
||||
func toErr(ctx context.Context, err error) error {
|
||||
if err == nil {
|
||||
return nil
|
||||
|
@@ -769,10 +769,13 @@ func (w *watchGrpcStream) joinSubstreams() {
|
||||
}
|
||||
}
|
||||
|
||||
var maxBackoff = 100 * time.Millisecond
|
||||
|
||||
// openWatchClient retries opening a watch client until success or halt.
|
||||
// manually retry in case "ws==nil && err==nil"
|
||||
// TODO: remove FailFast=false
|
||||
func (w *watchGrpcStream) openWatchClient() (ws pb.Watch_WatchClient, err error) {
|
||||
backoff := time.Millisecond
|
||||
for {
|
||||
select {
|
||||
case <-w.ctx.Done():
|
||||
@@ -788,6 +791,17 @@ func (w *watchGrpcStream) openWatchClient() (ws pb.Watch_WatchClient, err error)
|
||||
if isHaltErr(w.ctx, err) {
|
||||
return nil, v3rpc.Error(err)
|
||||
}
|
||||
if isUnavailableErr(w.ctx, err) {
|
||||
// retry, but backoff
|
||||
if backoff < maxBackoff {
|
||||
// 25% backoff factor
|
||||
backoff = backoff + backoff/4
|
||||
if backoff > maxBackoff {
|
||||
backoff = maxBackoff
|
||||
}
|
||||
}
|
||||
time.Sleep(backoff)
|
||||
}
|
||||
}
|
||||
return ws, nil
|
||||
}
|
||||
|
@@ -179,6 +179,8 @@ type etcdProcessClusterConfig struct {
|
||||
initialToken string
|
||||
quotaBackendBytes int64
|
||||
noStrictReconfig bool
|
||||
|
||||
cipherSuites []string
|
||||
}
|
||||
|
||||
// newEtcdProcessCluster launches a new cluster from etcd processes, returning
|
||||
@@ -384,6 +386,11 @@ func (cfg *etcdProcessClusterConfig) tlsArgs() (args []string) {
|
||||
args = append(args, tlsPeerArgs...)
|
||||
}
|
||||
}
|
||||
|
||||
if len(cfg.cipherSuites) > 0 {
|
||||
args = append(args, "--cipher-suites", strings.Join(cfg.cipherSuites, ","))
|
||||
}
|
||||
|
||||
return args
|
||||
}
|
||||
|
||||
|
@@ -127,6 +127,8 @@ type cURLReq struct {
|
||||
|
||||
value string
|
||||
expected string
|
||||
|
||||
ciphers string
|
||||
}
|
||||
|
||||
// cURLPrefixArgs builds the beginning of a curl command for a given key
|
||||
@@ -156,6 +158,10 @@ func cURLPrefixArgs(clus *etcdProcessCluster, method string, req cURLReq) []stri
|
||||
cmdArgs = append(cmdArgs, "-m", fmt.Sprintf("%d", req.timeout))
|
||||
}
|
||||
|
||||
if req.ciphers != "" {
|
||||
cmdArgs = append(cmdArgs, "--ciphers", req.ciphers)
|
||||
}
|
||||
|
||||
switch method {
|
||||
case "POST", "PUT":
|
||||
dt := req.value
|
||||
|
@@ -20,6 +20,7 @@ import (
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -27,6 +28,7 @@ import (
|
||||
"github.com/coreos/etcd/pkg/cors"
|
||||
"github.com/coreos/etcd/pkg/netutil"
|
||||
"github.com/coreos/etcd/pkg/srv"
|
||||
"github.com/coreos/etcd/pkg/tlsutil"
|
||||
"github.com/coreos/etcd/pkg/transport"
|
||||
"github.com/coreos/etcd/pkg/types"
|
||||
|
||||
@@ -158,6 +160,11 @@ type Config struct {
|
||||
PeerTLSInfo transport.TLSInfo
|
||||
PeerAutoTLS bool
|
||||
|
||||
// CipherSuites is a list of supported TLS cipher suites between
|
||||
// client/server and peers. If empty, Go auto-populates the list.
|
||||
// Note that cipher suites are prioritized in the given order.
|
||||
CipherSuites []string `json:"cipher-suites"`
|
||||
|
||||
// debug
|
||||
|
||||
Debug bool `json:"debug"`
|
||||
@@ -329,6 +336,25 @@ func (cfg *configYAML) configFromFile(path string) error {
|
||||
return cfg.Validate()
|
||||
}
|
||||
|
||||
func updateCipherSuites(tls *transport.TLSInfo, ss []string) error {
|
||||
if len(tls.CipherSuites) > 0 && len(ss) > 0 {
|
||||
return fmt.Errorf("TLSInfo.CipherSuites is already specified (given %v)", ss)
|
||||
}
|
||||
if len(ss) > 0 {
|
||||
cs := make([]uint16, len(ss))
|
||||
for i, s := range ss {
|
||||
var ok bool
|
||||
cs[i], ok = tlsutil.GetCipherSuite(s)
|
||||
if !ok {
|
||||
return fmt.Errorf("unexpected TLS cipher suite %q", s)
|
||||
}
|
||||
}
|
||||
tls.CipherSuites = cs
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Validate ensures that '*embed.Config' fields are properly configured.
|
||||
func (cfg *Config) Validate() error {
|
||||
if err := checkBindURLs(cfg.LPUrls); err != nil {
|
||||
return err
|
||||
@@ -431,6 +457,44 @@ func (cfg Config) defaultClientHost() bool {
|
||||
return len(cfg.ACUrls) == 1 && cfg.ACUrls[0].String() == DefaultAdvertiseClientURLs
|
||||
}
|
||||
|
||||
func (cfg *Config) ClientSelfCert() (err error) {
|
||||
if !cfg.ClientAutoTLS {
|
||||
return nil
|
||||
}
|
||||
if !cfg.ClientTLSInfo.Empty() {
|
||||
plog.Warningf("ignoring client auto TLS since certs given")
|
||||
return nil
|
||||
}
|
||||
chosts := make([]string, len(cfg.LCUrls))
|
||||
for i, u := range cfg.LCUrls {
|
||||
chosts[i] = u.Host
|
||||
}
|
||||
cfg.ClientTLSInfo, err = transport.SelfCert(filepath.Join(cfg.Dir, "fixtures", "client"), chosts)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return updateCipherSuites(&cfg.ClientTLSInfo, cfg.CipherSuites)
|
||||
}
|
||||
|
||||
func (cfg *Config) PeerSelfCert() (err error) {
|
||||
if !cfg.PeerAutoTLS {
|
||||
return nil
|
||||
}
|
||||
if !cfg.PeerTLSInfo.Empty() {
|
||||
plog.Warningf("ignoring peer auto TLS since certs given")
|
||||
return nil
|
||||
}
|
||||
phosts := make([]string, len(cfg.LPUrls))
|
||||
for i, u := range cfg.LPUrls {
|
||||
phosts[i] = u.Host
|
||||
}
|
||||
cfg.PeerTLSInfo, err = transport.SelfCert(filepath.Join(cfg.Dir, "fixtures", "peer"), phosts)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return updateCipherSuites(&cfg.PeerTLSInfo, cfg.CipherSuites)
|
||||
}
|
||||
|
||||
// UpdateDefaultClusterFromName updates cluster advertise URLs with, if available, default host,
|
||||
// if advertise URLs are default values(localhost:2379,2380) AND if listen URL is 0.0.0.0.
|
||||
// e.g. advertise peer URL localhost:2380 or listen peer URL 0.0.0.0:2380
|
||||
|
@@ -22,7 +22,6 @@ import (
|
||||
defaultLog "log"
|
||||
"net"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -264,17 +263,11 @@ func stopServers(ctx context.Context, ss *servers) {
|
||||
func (e *Etcd) Err() <-chan error { return e.errc }
|
||||
|
||||
func startPeerListeners(cfg *Config) (peers []*peerListener, err error) {
|
||||
if cfg.PeerAutoTLS && cfg.PeerTLSInfo.Empty() {
|
||||
phosts := make([]string, len(cfg.LPUrls))
|
||||
for i, u := range cfg.LPUrls {
|
||||
phosts[i] = u.Host
|
||||
}
|
||||
cfg.PeerTLSInfo, err = transport.SelfCert(filepath.Join(cfg.Dir, "fixtures", "peer"), phosts)
|
||||
if err != nil {
|
||||
plog.Fatalf("could not get certs (%v)", err)
|
||||
}
|
||||
} else if cfg.PeerAutoTLS {
|
||||
plog.Warningf("ignoring peer auto TLS since certs given")
|
||||
if err = updateCipherSuites(&cfg.PeerTLSInfo, cfg.CipherSuites); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err = cfg.PeerSelfCert(); err != nil {
|
||||
plog.Fatalf("could not get certs (%v)", err)
|
||||
}
|
||||
|
||||
if !cfg.PeerTLSInfo.Empty() {
|
||||
@@ -359,17 +352,11 @@ func (e *Etcd) servePeers() (err error) {
|
||||
}
|
||||
|
||||
func startClientListeners(cfg *Config) (sctxs map[string]*serveCtx, err error) {
|
||||
if cfg.ClientAutoTLS && cfg.ClientTLSInfo.Empty() {
|
||||
chosts := make([]string, len(cfg.LCUrls))
|
||||
for i, u := range cfg.LCUrls {
|
||||
chosts[i] = u.Host
|
||||
}
|
||||
cfg.ClientTLSInfo, err = transport.SelfCert(filepath.Join(cfg.Dir, "fixtures", "client"), chosts)
|
||||
if err != nil {
|
||||
plog.Fatalf("could not get certs (%v)", err)
|
||||
}
|
||||
} else if cfg.ClientAutoTLS {
|
||||
plog.Warningf("ignoring client auto TLS since certs given")
|
||||
if err = updateCipherSuites(&cfg.ClientTLSInfo, cfg.CipherSuites); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err = cfg.ClientSelfCert(); err != nil {
|
||||
plog.Fatalf("could not get certs (%v)", err)
|
||||
}
|
||||
|
||||
if cfg.EnablePprof {
|
||||
|
@@ -191,6 +191,8 @@ func newConfig() *config {
|
||||
fs.StringVar(&cfg.PeerTLSInfo.TrustedCAFile, "peer-trusted-ca-file", "", "Path to the peer server TLS trusted CA file.")
|
||||
fs.BoolVar(&cfg.PeerAutoTLS, "peer-auto-tls", false, "Peer TLS using generated certificates")
|
||||
|
||||
fs.Var(flags.NewStringsValueV2(""), "cipher-suites", "Comma-separated list of supported TLS cipher suites between client/server and peers (empty will be auto-populated by Go).")
|
||||
|
||||
// logging
|
||||
fs.BoolVar(&cfg.Debug, "debug", false, "Enable debug-level logging for etcd.")
|
||||
fs.StringVar(&cfg.LogPkgLevels, "log-package-levels", "", "Specify a particular log level for each etcd package (eg: 'etcdmain=CRITICAL,etcdserver=DEBUG').")
|
||||
@@ -266,6 +268,8 @@ func (cfg *config) configFromCmdLine() error {
|
||||
cfg.Fallback = cfg.fallback.String()
|
||||
cfg.Proxy = cfg.proxy.String()
|
||||
|
||||
cfg.CipherSuites = flags.StringsFromFlagV2(cfg.FlagSet, "cipher-suites")
|
||||
|
||||
// disable default advertise-client-urls if lcurls is set
|
||||
missingAC := flags.IsSet(cfg.FlagSet, "listen-client-urls") && !flags.IsSet(cfg.FlagSet, "advertise-client-urls")
|
||||
if !cfg.mayBeProxy() && missingAC {
|
||||
|
@@ -150,6 +150,8 @@ security flags:
|
||||
path to the peer server TLS trusted CA file.
|
||||
--peer-auto-tls 'false'
|
||||
peer TLS using self-generated certificates if --peer-key-file and --peer-cert-file are not provided.
|
||||
--cipher-suites ''
|
||||
comma-separated list of supported TLS cipher suites between client/server and peers (empty will be auto-populated by Go).
|
||||
|
||||
logging flags
|
||||
|
||||
|
@@ -89,6 +89,9 @@ func (s *EtcdServer) newApplierV3() applierV3 {
|
||||
|
||||
func (a *applierV3backend) Apply(r *pb.InternalRaftRequest) *applyResult {
|
||||
ar := &applyResult{}
|
||||
defer func(start time.Time) {
|
||||
warnOfExpensiveRequest(start, &pb.InternalRaftStringer{Request: r}, ar.resp, ar.err)
|
||||
}(time.Now())
|
||||
|
||||
// call into a.s.applyV3.F instead of a.F so upper appliers can check individual calls
|
||||
switch {
|
||||
|
@@ -105,10 +105,12 @@ func (a *applierV2store) Sync(r *pb.Request) Response {
|
||||
return Response{}
|
||||
}
|
||||
|
||||
// applyV2Request interprets r as a call to store.X and returns a Response interpreted
|
||||
// from store.Event
|
||||
// applyV2Request interprets r as a call to v2store.X
|
||||
// and returns a Response interpreted from v2store.Event
|
||||
func (s *EtcdServer) applyV2Request(r *pb.Request) Response {
|
||||
defer warnOfExpensiveRequest(time.Now(), r, nil, nil)
|
||||
toTTLOptions(r)
|
||||
|
||||
switch r.Method {
|
||||
case "POST":
|
||||
return s.applyV2.Post(r)
|
||||
|
179
etcdserver/etcdserverpb/raft_internal_stringer.go
Normal file
179
etcdserver/etcdserverpb/raft_internal_stringer.go
Normal file
@@ -0,0 +1,179 @@
|
||||
// Copyright 2018 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 etcdserverpb
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
proto "github.com/golang/protobuf/proto"
|
||||
)
|
||||
|
||||
// InternalRaftStringer implements custom proto Stringer:
|
||||
// redact password, replace value fields with value_size fields.
|
||||
type InternalRaftStringer struct {
|
||||
Request *InternalRaftRequest
|
||||
}
|
||||
|
||||
func (as *InternalRaftStringer) String() string {
|
||||
switch {
|
||||
case as.Request.LeaseGrant != nil:
|
||||
return fmt.Sprintf("header:<%s> lease_grant:<ttl:%d-second id:%016x>",
|
||||
as.Request.Header.String(),
|
||||
as.Request.LeaseGrant.TTL,
|
||||
as.Request.LeaseGrant.ID,
|
||||
)
|
||||
case as.Request.LeaseRevoke != nil:
|
||||
return fmt.Sprintf("header:<%s> lease_revoke:<id:%016x>",
|
||||
as.Request.Header.String(),
|
||||
as.Request.LeaseRevoke.ID,
|
||||
)
|
||||
case as.Request.Authenticate != nil:
|
||||
return fmt.Sprintf("header:<%s> authenticate:<name:%s simple_token:%s>",
|
||||
as.Request.Header.String(),
|
||||
as.Request.Authenticate.Name,
|
||||
as.Request.Authenticate.SimpleToken,
|
||||
)
|
||||
case as.Request.AuthUserAdd != nil:
|
||||
return fmt.Sprintf("header:<%s> auth_user_add:<name:%s>",
|
||||
as.Request.Header.String(),
|
||||
as.Request.AuthUserAdd.Name,
|
||||
)
|
||||
case as.Request.AuthUserChangePassword != nil:
|
||||
return fmt.Sprintf("header:<%s> auth_user_change_password:<name:%s>",
|
||||
as.Request.Header.String(),
|
||||
as.Request.AuthUserChangePassword.Name,
|
||||
)
|
||||
case as.Request.Put != nil:
|
||||
return fmt.Sprintf("header:<%s> put:<%s>",
|
||||
as.Request.Header.String(),
|
||||
newLoggablePutRequest(as.Request.Put).String(),
|
||||
)
|
||||
case as.Request.Txn != nil:
|
||||
return fmt.Sprintf("header:<%s> txn:<%s>",
|
||||
as.Request.Header.String(),
|
||||
NewLoggableTxnRequest(as.Request.Txn).String(),
|
||||
)
|
||||
default:
|
||||
// nothing to redact
|
||||
}
|
||||
return as.Request.String()
|
||||
}
|
||||
|
||||
// txnRequestStringer implements a custom proto String to replace value bytes fields with value size
|
||||
// fields in any nested txn and put operations.
|
||||
type txnRequestStringer struct {
|
||||
Request *TxnRequest
|
||||
}
|
||||
|
||||
func NewLoggableTxnRequest(request *TxnRequest) *txnRequestStringer {
|
||||
return &txnRequestStringer{request}
|
||||
}
|
||||
|
||||
func (as *txnRequestStringer) String() string {
|
||||
var compare []string
|
||||
for _, c := range as.Request.Compare {
|
||||
switch cv := c.TargetUnion.(type) {
|
||||
case *Compare_Value:
|
||||
compare = append(compare, newLoggableValueCompare(c, cv).String())
|
||||
default:
|
||||
// nothing to redact
|
||||
compare = append(compare, c.String())
|
||||
}
|
||||
}
|
||||
var success []string
|
||||
for _, s := range as.Request.Success {
|
||||
success = append(success, newLoggableRequestOp(s).String())
|
||||
}
|
||||
var failure []string
|
||||
for _, f := range as.Request.Failure {
|
||||
failure = append(failure, newLoggableRequestOp(f).String())
|
||||
}
|
||||
return fmt.Sprintf("compare:<%s> success:<%s> failure:<%s>",
|
||||
strings.Join(compare, " "),
|
||||
strings.Join(success, " "),
|
||||
strings.Join(failure, " "),
|
||||
)
|
||||
}
|
||||
|
||||
// requestOpStringer implements a custom proto String to replace value bytes fields with value
|
||||
// size fields in any nested txn and put operations.
|
||||
type requestOpStringer struct {
|
||||
Op *RequestOp
|
||||
}
|
||||
|
||||
func newLoggableRequestOp(op *RequestOp) *requestOpStringer {
|
||||
return &requestOpStringer{op}
|
||||
}
|
||||
|
||||
func (as *requestOpStringer) String() string {
|
||||
switch op := as.Op.Request.(type) {
|
||||
case *RequestOp_RequestPut:
|
||||
return fmt.Sprintf("request_put:<%s>", newLoggablePutRequest(op.RequestPut).String())
|
||||
default:
|
||||
// nothing to redact
|
||||
}
|
||||
return as.Op.String()
|
||||
}
|
||||
|
||||
// loggableValueCompare implements a custom proto String for Compare.Value union member types to
|
||||
// replace the value bytes field with a value size field.
|
||||
// To preserve proto encoding of the key and range_end bytes, a faked out proto type is used here.
|
||||
type loggableValueCompare struct {
|
||||
Result Compare_CompareResult `protobuf:"varint,1,opt,name=result,proto3,enum=etcdserverpb.Compare_CompareResult"`
|
||||
Target Compare_CompareTarget `protobuf:"varint,2,opt,name=target,proto3,enum=etcdserverpb.Compare_CompareTarget"`
|
||||
Key []byte `protobuf:"bytes,3,opt,name=key,proto3"`
|
||||
ValueSize int `protobuf:"bytes,7,opt,name=value_size,proto3"`
|
||||
}
|
||||
|
||||
func newLoggableValueCompare(c *Compare, cv *Compare_Value) *loggableValueCompare {
|
||||
return &loggableValueCompare{
|
||||
c.Result,
|
||||
c.Target,
|
||||
c.Key,
|
||||
len(cv.Value),
|
||||
}
|
||||
}
|
||||
|
||||
func (m *loggableValueCompare) Reset() { *m = loggableValueCompare{} }
|
||||
func (m *loggableValueCompare) String() string { return proto.CompactTextString(m) }
|
||||
func (*loggableValueCompare) ProtoMessage() {}
|
||||
|
||||
// loggablePutRequest implements a custom proto String to replace value bytes field with a value
|
||||
// size field.
|
||||
// To preserve proto encoding of the key bytes, a faked out proto type is used here.
|
||||
type loggablePutRequest struct {
|
||||
Key []byte `protobuf:"bytes,1,opt,name=key,proto3"`
|
||||
ValueSize int `protobuf:"varint,2,opt,name=value_size,proto3"`
|
||||
Lease int64 `protobuf:"varint,3,opt,name=lease,proto3"`
|
||||
PrevKv bool `protobuf:"varint,4,opt,name=prev_kv,proto3"`
|
||||
IgnoreValue bool `protobuf:"varint,5,opt,name=ignore_value,proto3"`
|
||||
IgnoreLease bool `protobuf:"varint,6,opt,name=ignore_lease,proto3"`
|
||||
}
|
||||
|
||||
func newLoggablePutRequest(request *PutRequest) *loggablePutRequest {
|
||||
return &loggablePutRequest{
|
||||
request.Key,
|
||||
len(request.Value),
|
||||
request.Lease,
|
||||
request.PrevKv,
|
||||
request.IgnoreValue,
|
||||
request.IgnoreLease,
|
||||
}
|
||||
}
|
||||
|
||||
func (m *loggablePutRequest) Reset() { *m = loggablePutRequest{} }
|
||||
func (m *loggablePutRequest) String() string { return proto.CompactTextString(m) }
|
||||
func (*loggablePutRequest) ProtoMessage() {}
|
@@ -18,6 +18,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/coreos/etcd/pkg/runtime"
|
||||
"github.com/coreos/etcd/version"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
)
|
||||
|
||||
@@ -70,6 +71,13 @@ var (
|
||||
Name: "lease_expired_total",
|
||||
Help: "The total number of expired leases.",
|
||||
})
|
||||
currentVersion = prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
||||
Namespace: "etcd",
|
||||
Subsystem: "server",
|
||||
Name: "version",
|
||||
Help: "Which version is running. 1 for 'server_version' label with current version.",
|
||||
},
|
||||
[]string{"server_version"})
|
||||
)
|
||||
|
||||
func init() {
|
||||
@@ -81,6 +89,11 @@ func init() {
|
||||
prometheus.MustRegister(proposalsPending)
|
||||
prometheus.MustRegister(proposalsFailed)
|
||||
prometheus.MustRegister(leaseExpired)
|
||||
prometheus.MustRegister(currentVersion)
|
||||
|
||||
currentVersion.With(prometheus.Labels{
|
||||
"server_version": version.Version,
|
||||
}).Set(1)
|
||||
}
|
||||
|
||||
func monitorFileDescriptor(done <-chan struct{}) {
|
||||
|
@@ -597,18 +597,21 @@ func (s *EtcdServer) start() {
|
||||
}
|
||||
|
||||
func (s *EtcdServer) purgeFile() {
|
||||
var serrc, werrc <-chan error
|
||||
var dberrc, serrc, werrc <-chan error
|
||||
if s.Cfg.MaxSnapFiles > 0 {
|
||||
dberrc = fileutil.PurgeFile(s.Cfg.SnapDir(), "snap.db", s.Cfg.MaxSnapFiles, purgeFileInterval, s.done)
|
||||
serrc = fileutil.PurgeFile(s.Cfg.SnapDir(), "snap", s.Cfg.MaxSnapFiles, purgeFileInterval, s.done)
|
||||
}
|
||||
if s.Cfg.MaxWALFiles > 0 {
|
||||
werrc = fileutil.PurgeFile(s.Cfg.WALDir(), "wal", s.Cfg.MaxWALFiles, purgeFileInterval, s.done)
|
||||
}
|
||||
select {
|
||||
case e := <-werrc:
|
||||
plog.Fatalf("failed to purge wal file %v", e)
|
||||
case e := <-dberrc:
|
||||
plog.Fatalf("failed to purge snap db file %v", e)
|
||||
case e := <-serrc:
|
||||
plog.Fatalf("failed to purge snap file %v", e)
|
||||
case e := <-werrc:
|
||||
plog.Fatalf("failed to purge wal file %v", e)
|
||||
case <-s.stopping:
|
||||
return
|
||||
}
|
||||
@@ -815,14 +818,8 @@ func (s *EtcdServer) run() {
|
||||
|
||||
func (s *EtcdServer) applyAll(ep *etcdProgress, apply *apply) {
|
||||
s.applySnapshot(ep, apply)
|
||||
st := time.Now()
|
||||
s.applyEntries(ep, apply)
|
||||
d := time.Since(st)
|
||||
entriesNum := len(apply.entries)
|
||||
if entriesNum != 0 && d > time.Duration(entriesNum)*warnApplyDuration {
|
||||
plog.Warningf("apply entries took too long [%v for %d entries]", d, len(apply.entries))
|
||||
plog.Warningf("avoid queries with large range/delete range!")
|
||||
}
|
||||
|
||||
proposalsApplied.Set(float64(ep.appliedi))
|
||||
s.applyWait.Trigger(ep.appliedi)
|
||||
// wait for the raft routine to finish the disk writes before triggering a
|
||||
|
@@ -15,11 +15,16 @@
|
||||
package etcdserver
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"reflect"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
|
||||
"github.com/coreos/etcd/etcdserver/membership"
|
||||
"github.com/coreos/etcd/pkg/types"
|
||||
"github.com/coreos/etcd/rafthttp"
|
||||
"github.com/golang/protobuf/proto"
|
||||
)
|
||||
|
||||
// isConnectedToQuorumSince checks whether the local member is connected to the
|
||||
@@ -95,3 +100,55 @@ func (nc *notifier) notify(err error) {
|
||||
nc.err = err
|
||||
close(nc.c)
|
||||
}
|
||||
|
||||
func warnOfExpensiveRequest(now time.Time, reqStringer fmt.Stringer, respMsg proto.Message, err error) {
|
||||
var resp string
|
||||
if !isNil(respMsg) {
|
||||
resp = fmt.Sprintf("size:%d", proto.Size(respMsg))
|
||||
}
|
||||
warnOfExpensiveGenericRequest(now, reqStringer, "", resp, err)
|
||||
}
|
||||
|
||||
func warnOfExpensiveReadOnlyTxnRequest(now time.Time, r *pb.TxnRequest, txnResponse *pb.TxnResponse, err error) {
|
||||
reqStringer := pb.NewLoggableTxnRequest(r)
|
||||
var resp string
|
||||
if !isNil(txnResponse) {
|
||||
var resps []string
|
||||
for _, r := range txnResponse.Responses {
|
||||
switch op := r.Response.(type) {
|
||||
case *pb.ResponseOp_ResponseRange:
|
||||
resps = append(resps, fmt.Sprintf("range_response_count:%d", len(op.ResponseRange.Kvs)))
|
||||
default:
|
||||
// only range responses should be in a read only txn request
|
||||
}
|
||||
}
|
||||
resp = fmt.Sprintf("responses:<%s> size:%d", strings.Join(resps, " "), proto.Size(txnResponse))
|
||||
}
|
||||
warnOfExpensiveGenericRequest(now, reqStringer, "read-only range ", resp, err)
|
||||
}
|
||||
|
||||
func warnOfExpensiveReadOnlyRangeRequest(now time.Time, reqStringer fmt.Stringer, rangeResponse *pb.RangeResponse, err error) {
|
||||
var resp string
|
||||
if !isNil(rangeResponse) {
|
||||
resp = fmt.Sprintf("range_response_count:%d size:%d", len(rangeResponse.Kvs), proto.Size(rangeResponse))
|
||||
}
|
||||
warnOfExpensiveGenericRequest(now, reqStringer, "read-only range ", resp, err)
|
||||
}
|
||||
|
||||
func warnOfExpensiveGenericRequest(now time.Time, reqStringer fmt.Stringer, prefix string, resp string, err error) {
|
||||
// TODO: add metrics
|
||||
d := time.Since(now)
|
||||
if d > warnApplyDuration {
|
||||
var result string
|
||||
if err != nil {
|
||||
result = fmt.Sprintf("error:%v", err)
|
||||
} else {
|
||||
result = resp
|
||||
}
|
||||
plog.Warningf("%srequest %q with result %q took too long (%v) to execute", prefix, reqStringer.String(), result, d)
|
||||
}
|
||||
}
|
||||
|
||||
func isNil(msg proto.Message) bool {
|
||||
return msg == nil || reflect.ValueOf(msg).IsNil()
|
||||
}
|
||||
|
@@ -19,8 +19,6 @@ import (
|
||||
"encoding/binary"
|
||||
"time"
|
||||
|
||||
"github.com/gogo/protobuf/proto"
|
||||
|
||||
"github.com/coreos/etcd/auth"
|
||||
pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
|
||||
"github.com/coreos/etcd/etcdserver/membership"
|
||||
@@ -28,7 +26,7 @@ import (
|
||||
"github.com/coreos/etcd/lease/leasehttp"
|
||||
"github.com/coreos/etcd/mvcc"
|
||||
"github.com/coreos/etcd/raft"
|
||||
|
||||
"github.com/gogo/protobuf/proto"
|
||||
"golang.org/x/net/context"
|
||||
)
|
||||
|
||||
@@ -82,20 +80,26 @@ type Authenticator interface {
|
||||
}
|
||||
|
||||
func (s *EtcdServer) Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeResponse, error) {
|
||||
var resp *pb.RangeResponse
|
||||
var err error
|
||||
defer func(start time.Time) {
|
||||
warnOfExpensiveReadOnlyRangeRequest(start, r, resp, err)
|
||||
}(time.Now())
|
||||
|
||||
if !r.Serializable {
|
||||
err := s.linearizableReadNotify(ctx)
|
||||
err = s.linearizableReadNotify(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
var resp *pb.RangeResponse
|
||||
var err error
|
||||
chk := func(ai *auth.AuthInfo) error {
|
||||
return s.authStore.IsRangePermitted(ai, r.Key, r.RangeEnd)
|
||||
}
|
||||
|
||||
get := func() { resp, err = s.applyV3Base.Range(nil, r) }
|
||||
if serr := s.doSerialize(ctx, chk, get); serr != nil {
|
||||
return nil, serr
|
||||
err = serr
|
||||
return nil, err
|
||||
}
|
||||
return resp, err
|
||||
}
|
||||
@@ -129,12 +133,18 @@ func (s *EtcdServer) Txn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse
|
||||
chk := func(ai *auth.AuthInfo) error {
|
||||
return checkTxnAuth(s.authStore, ai, r)
|
||||
}
|
||||
|
||||
defer func(start time.Time) {
|
||||
warnOfExpensiveReadOnlyTxnRequest(start, r, resp, err)
|
||||
}(time.Now())
|
||||
|
||||
get := func() { resp, err = s.applyV3Base.Txn(r) }
|
||||
if serr := s.doSerialize(ctx, chk, get); serr != nil {
|
||||
return nil, serr
|
||||
}
|
||||
return resp, err
|
||||
}
|
||||
|
||||
resp, err := s.raftRequest(ctx, pb.InternalRaftRequest{Txn: r})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
71
integration/v3_tls_test.go
Normal file
71
integration/v3_tls_test.go
Normal file
@@ -0,0 +1,71 @@
|
||||
// Copyright 2018 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 integration
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/coreos/etcd/clientv3"
|
||||
"github.com/coreos/etcd/pkg/testutil"
|
||||
)
|
||||
|
||||
func TestTLSClientCipherSuitesValid(t *testing.T) { testTLSCipherSuites(t, true) }
|
||||
func TestTLSClientCipherSuitesMismatch(t *testing.T) { testTLSCipherSuites(t, false) }
|
||||
|
||||
// testTLSCipherSuites ensures mismatching client-side cipher suite
|
||||
// fail TLS handshake with the server.
|
||||
func testTLSCipherSuites(t *testing.T, valid bool) {
|
||||
defer testutil.AfterTest(t)
|
||||
|
||||
cipherSuites := []uint16{
|
||||
tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
|
||||
tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
|
||||
tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
|
||||
tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
|
||||
tls.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305,
|
||||
tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305,
|
||||
}
|
||||
srvTLS, cliTLS := testTLSInfo, testTLSInfo
|
||||
if valid {
|
||||
srvTLS.CipherSuites, cliTLS.CipherSuites = cipherSuites, cipherSuites
|
||||
} else {
|
||||
srvTLS.CipherSuites, cliTLS.CipherSuites = cipherSuites[:2], cipherSuites[2:]
|
||||
}
|
||||
|
||||
clus := NewClusterV3(t, &ClusterConfig{Size: 1, ClientTLS: &srvTLS})
|
||||
defer clus.Terminate(t)
|
||||
|
||||
cc, err := cliTLS.ClientConfig()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cli, cerr := clientv3.New(clientv3.Config{
|
||||
Endpoints: []string{clus.Members[0].GRPCAddr()},
|
||||
DialTimeout: time.Second,
|
||||
TLS: cc,
|
||||
})
|
||||
if cli != nil {
|
||||
cli.Close()
|
||||
}
|
||||
if !valid && cerr != context.DeadlineExceeded {
|
||||
t.Fatalf("expected %v with TLS handshake failure, got %v", context.DeadlineExceeded, cerr)
|
||||
}
|
||||
if valid && cerr != nil {
|
||||
t.Fatalf("expected TLS handshake success, got %v", cerr)
|
||||
}
|
||||
}
|
@@ -188,6 +188,7 @@ func (s *watchableStore) Restore(b backend.Backend) error {
|
||||
}
|
||||
|
||||
for wa := range s.synced.watchers {
|
||||
wa.restore = true
|
||||
s.unsynced.add(wa)
|
||||
}
|
||||
s.synced = newWatcherGroup()
|
||||
@@ -479,6 +480,14 @@ type watcher struct {
|
||||
// compacted is set when the watcher is removed because of compaction
|
||||
compacted bool
|
||||
|
||||
// restore is true when the watcher is being restored from leader snapshot
|
||||
// which means that this watcher has just been moved from "synced" to "unsynced"
|
||||
// watcher group, possibly with a future revision when it was first added
|
||||
// to the synced watcher
|
||||
// "unsynced" watcher revision must always be <= current revision,
|
||||
// except when the watcher were to be moved from "synced" watcher group
|
||||
restore bool
|
||||
|
||||
// minRev is the minimum revision update the watcher will accept
|
||||
minRev int64
|
||||
id WatchID
|
||||
|
@@ -336,6 +336,62 @@ func TestWatchRestore(t *testing.T) {
|
||||
t.Run("RunSyncWatchLoopBeforeRestore", test(time.Millisecond*120)) // longer than default waitDuration
|
||||
}
|
||||
|
||||
// TestWatchRestoreSyncedWatcher tests such a case that:
|
||||
// 1. watcher is created with a future revision "math.MaxInt64 - 2"
|
||||
// 2. watcher with a future revision is added to "synced" watcher group
|
||||
// 3. restore/overwrite storage with snapshot of a higher lasat revision
|
||||
// 4. restore operation moves "synced" to "unsynced" watcher group
|
||||
// 5. choose the watcher from step 1, without panic
|
||||
func TestWatchRestoreSyncedWatcher(t *testing.T) {
|
||||
b1, b1Path := backend.NewDefaultTmpBackend()
|
||||
s1 := newWatchableStore(b1, &lease.FakeLessor{}, nil)
|
||||
defer cleanup(s1, b1, b1Path)
|
||||
|
||||
b2, b2Path := backend.NewDefaultTmpBackend()
|
||||
s2 := newWatchableStore(b2, &lease.FakeLessor{}, nil)
|
||||
defer cleanup(s2, b2, b2Path)
|
||||
|
||||
testKey, testValue := []byte("foo"), []byte("bar")
|
||||
rev := s1.Put(testKey, testValue, lease.NoLease)
|
||||
startRev := rev + 2
|
||||
|
||||
// create a watcher with a future revision
|
||||
// add to "synced" watcher group (startRev > s.store.currentRev)
|
||||
w1 := s1.NewWatchStream()
|
||||
w1.Watch(testKey, nil, startRev)
|
||||
|
||||
// make "s2" ends up with a higher last revision
|
||||
s2.Put(testKey, testValue, lease.NoLease)
|
||||
s2.Put(testKey, testValue, lease.NoLease)
|
||||
|
||||
// overwrite storage with higher revisions
|
||||
if err := s1.Restore(b2); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// wait for next "syncWatchersLoop" iteration
|
||||
// and the unsynced watcher should be chosen
|
||||
time.Sleep(2 * time.Second)
|
||||
|
||||
// trigger events for "startRev"
|
||||
s1.Put(testKey, testValue, lease.NoLease)
|
||||
|
||||
select {
|
||||
case resp := <-w1.Chan():
|
||||
if resp.Revision != startRev {
|
||||
t.Fatalf("resp.Revision expect %d, got %d", startRev, resp.Revision)
|
||||
}
|
||||
if len(resp.Events) != 1 {
|
||||
t.Fatalf("len(resp.Events) expect 1, got %d", len(resp.Events))
|
||||
}
|
||||
if resp.Events[0].Kv.ModRevision != startRev {
|
||||
t.Fatalf("resp.Events[0].Kv.ModRevision expect %d, got %d", startRev, resp.Events[0].Kv.ModRevision)
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("failed to receive event in 1 second")
|
||||
}
|
||||
}
|
||||
|
||||
// TestWatchBatchUnsynced tests batching on unsynced watchers
|
||||
func TestWatchBatchUnsynced(t *testing.T) {
|
||||
b, tmpPath := backend.NewDefaultTmpBackend()
|
||||
|
@@ -15,6 +15,7 @@
|
||||
package mvcc
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math"
|
||||
|
||||
"github.com/coreos/etcd/mvcc/mvccpb"
|
||||
@@ -238,7 +239,15 @@ func (wg *watcherGroup) chooseAll(curRev, compactRev int64) int64 {
|
||||
minRev := int64(math.MaxInt64)
|
||||
for w := range wg.watchers {
|
||||
if w.minRev > curRev {
|
||||
panic("watcher current revision should not exceed current revision")
|
||||
// after network partition, possibly choosing future revision watcher from restore operation
|
||||
// with watch key "proxy-namespace__lostleader" and revision "math.MaxInt64 - 2"
|
||||
// do not panic when such watcher had been moved from "synced" watcher during restore operation
|
||||
if !w.restore {
|
||||
panic(fmt.Errorf("watcher minimum revision %d should not exceed current revision %d", w.minRev, curRev))
|
||||
}
|
||||
|
||||
// mark 'restore' done, since it's chosen
|
||||
w.restore = false
|
||||
}
|
||||
if w.minRev < compactRev {
|
||||
select {
|
||||
|
@@ -14,7 +14,12 @@
|
||||
|
||||
package flags
|
||||
|
||||
import "errors"
|
||||
import (
|
||||
"errors"
|
||||
"flag"
|
||||
"sort"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// NewStringsFlag creates a new string flag for which any one of the given
|
||||
// strings is a valid value, and any other value is an error.
|
||||
@@ -44,3 +49,34 @@ func (ss *StringsFlag) Set(s string) error {
|
||||
func (ss *StringsFlag) String() string {
|
||||
return ss.val
|
||||
}
|
||||
|
||||
// StringsValueV2 wraps "sort.StringSlice".
|
||||
type StringsValueV2 sort.StringSlice
|
||||
|
||||
// Set parses a command line set of strings, separated by comma.
|
||||
// Implements "flag.Value" interface.
|
||||
func (ss *StringsValueV2) Set(s string) error {
|
||||
*ss = strings.Split(s, ",")
|
||||
return nil
|
||||
}
|
||||
|
||||
// String implements "flag.Value" interface.
|
||||
func (ss *StringsValueV2) String() string { return strings.Join(*ss, ",") }
|
||||
|
||||
// NewStringsValueV2 implements string slice as "flag.Value" interface.
|
||||
// Given value is to be separated by comma.
|
||||
func NewStringsValueV2(s string) (ss *StringsValueV2) {
|
||||
if s == "" {
|
||||
return &StringsValueV2{}
|
||||
}
|
||||
ss = new(StringsValueV2)
|
||||
if err := ss.Set(s); err != nil {
|
||||
plog.Panicf("new StringsValueV2 should never fail: %v", err)
|
||||
}
|
||||
return ss
|
||||
}
|
||||
|
||||
// StringsFromFlagV2 returns a string slice from the flag.
|
||||
func StringsFromFlagV2(fs *flag.FlagSet, flagName string) []string {
|
||||
return []string(*fs.Lookup(flagName).Value.(*StringsValueV2))
|
||||
}
|
||||
|
51
pkg/tlsutil/cipher_suites.go
Normal file
51
pkg/tlsutil/cipher_suites.go
Normal file
@@ -0,0 +1,51 @@
|
||||
// Copyright 2018 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 tlsutil
|
||||
|
||||
import "crypto/tls"
|
||||
|
||||
// cipher suites implemented by Go
|
||||
// https://github.com/golang/go/blob/dev.boringcrypto.go1.10/src/crypto/tls/cipher_suites.go
|
||||
var cipherSuites = map[string]uint16{
|
||||
"TLS_RSA_WITH_RC4_128_SHA": tls.TLS_RSA_WITH_RC4_128_SHA,
|
||||
"TLS_RSA_WITH_3DES_EDE_CBC_SHA": tls.TLS_RSA_WITH_3DES_EDE_CBC_SHA,
|
||||
"TLS_RSA_WITH_AES_128_CBC_SHA": tls.TLS_RSA_WITH_AES_128_CBC_SHA,
|
||||
"TLS_RSA_WITH_AES_256_CBC_SHA": tls.TLS_RSA_WITH_AES_256_CBC_SHA,
|
||||
"TLS_RSA_WITH_AES_128_CBC_SHA256": tls.TLS_RSA_WITH_AES_128_CBC_SHA256,
|
||||
"TLS_RSA_WITH_AES_128_GCM_SHA256": tls.TLS_RSA_WITH_AES_128_GCM_SHA256,
|
||||
"TLS_RSA_WITH_AES_256_GCM_SHA384": tls.TLS_RSA_WITH_AES_256_GCM_SHA384,
|
||||
"TLS_ECDHE_ECDSA_WITH_RC4_128_SHA": tls.TLS_ECDHE_ECDSA_WITH_RC4_128_SHA,
|
||||
"TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA": tls.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA,
|
||||
"TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA": tls.TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA,
|
||||
"TLS_ECDHE_RSA_WITH_RC4_128_SHA": tls.TLS_ECDHE_RSA_WITH_RC4_128_SHA,
|
||||
"TLS_ECDHE_RSA_WITH_3DES_EDE_CBC_SHA": tls.TLS_ECDHE_RSA_WITH_3DES_EDE_CBC_SHA,
|
||||
"TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA": tls.TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA,
|
||||
"TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA": tls.TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA,
|
||||
"TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256": tls.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256,
|
||||
"TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256": tls.TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256,
|
||||
"TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256": tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
|
||||
"TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256": tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
|
||||
"TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384": tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
|
||||
"TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384": tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
|
||||
"TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305": tls.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305,
|
||||
"TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305": tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305,
|
||||
}
|
||||
|
||||
// GetCipherSuite returns the corresponding cipher suite,
|
||||
// and boolean value if it is supported.
|
||||
func GetCipherSuite(s string) (uint16, bool) {
|
||||
v, ok := cipherSuites[s]
|
||||
return v, ok
|
||||
}
|
@@ -69,6 +69,11 @@ type TLSInfo struct {
|
||||
// connection will be closed immediately afterwards.
|
||||
HandshakeFailure func(*tls.Conn, error)
|
||||
|
||||
// CipherSuites is a list of supported cipher suites.
|
||||
// If empty, Go auto-populates it by default.
|
||||
// Note that cipher suites are prioritized in the given order.
|
||||
CipherSuites []uint16
|
||||
|
||||
selfCert bool
|
||||
|
||||
// parseFunc exists to simplify testing. Typically, parseFunc
|
||||
@@ -171,6 +176,11 @@ func (info TLSInfo) baseConfig() (*tls.Config, error) {
|
||||
MinVersion: tls.VersionTLS12,
|
||||
ServerName: info.ServerName,
|
||||
}
|
||||
|
||||
if len(info.CipherSuites) > 0 {
|
||||
cfg.CipherSuites = info.CipherSuites
|
||||
}
|
||||
|
||||
// this only reloads certs when there's a client request
|
||||
// TODO: support server-side refresh (e.g. inotify, SIGHUP), caching
|
||||
cfg.GetCertificate = func(clientHello *tls.ClientHelloInfo) (*tls.Certificate, error) {
|
||||
|
73
pkg/transport/transport_test.go
Normal file
73
pkg/transport/transport_test.go
Normal file
@@ -0,0 +1,73 @@
|
||||
// Copyright 2018 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 transport
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"net/http"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TestNewTransportTLSInvalidCipherSuites expects a client with invalid
|
||||
// cipher suites fail to handshake with the server.
|
||||
func TestNewTransportTLSInvalidCipherSuites(t *testing.T) {
|
||||
tlsInfo, del, err := createSelfCert()
|
||||
if err != nil {
|
||||
t.Fatalf("unable to create cert: %v", err)
|
||||
}
|
||||
defer del()
|
||||
|
||||
cipherSuites := []uint16{
|
||||
tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
|
||||
tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
|
||||
tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
|
||||
tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
|
||||
tls.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305,
|
||||
tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305,
|
||||
}
|
||||
|
||||
// make server and client have unmatched cipher suites
|
||||
srvTLS, cliTLS := *tlsInfo, *tlsInfo
|
||||
srvTLS.CipherSuites, cliTLS.CipherSuites = cipherSuites[:2], cipherSuites[2:]
|
||||
|
||||
ln, err := NewListener("127.0.0.1:0", "https", &srvTLS)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected NewListener error: %v", err)
|
||||
}
|
||||
defer ln.Close()
|
||||
|
||||
donec := make(chan struct{})
|
||||
go func() {
|
||||
ln.Accept()
|
||||
donec <- struct{}{}
|
||||
}()
|
||||
go func() {
|
||||
tr, err := NewTransport(cliTLS, 3*time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected NewTransport error: %v", err)
|
||||
}
|
||||
cli := &http.Client{Transport: tr}
|
||||
_, gerr := cli.Get("https://" + ln.Addr().String())
|
||||
if gerr == nil || !strings.Contains(gerr.Error(), "tls: handshake failure") {
|
||||
t.Fatal("expected client TLS handshake error")
|
||||
}
|
||||
ln.Close()
|
||||
donec <- struct{}{}
|
||||
}()
|
||||
<-donec
|
||||
<-donec
|
||||
}
|
@@ -5,17 +5,34 @@ if ! [[ "$0" =~ "tests/semaphore.test.bash" ]]; then
|
||||
exit 255
|
||||
fi
|
||||
|
||||
TEST_SUFFIX=$(date +%s | base64 | head -c 15)
|
||||
<<COMMENT
|
||||
# amd64-e2e
|
||||
bash tests/semaphore.test.bash
|
||||
|
||||
TEST_OPTS="PASSES='build release e2e' MANUAL_VER=v3.2.17"
|
||||
if [ "$TEST_ARCH" == "386" ]; then
|
||||
# 386-e2e
|
||||
TEST_ARCH=386 bash tests/semaphore.test.bash
|
||||
|
||||
# grpc-proxy
|
||||
TEST_OPTS="PASSES='build grpcproxy'" bash tests/semaphore.test.bash
|
||||
|
||||
# coverage
|
||||
TEST_OPTS="coverage" bash tests/semaphore.test.bash
|
||||
COMMENT
|
||||
|
||||
if [ -z "${TEST_OPTS}" ]; then
|
||||
TEST_OPTS="PASSES='build release e2e' MANUAL_VER=v3.2.22"
|
||||
fi
|
||||
if [ "${TEST_ARCH}" == "386" ]; then
|
||||
TEST_OPTS="GOARCH=386 PASSES='build e2e'"
|
||||
fi
|
||||
|
||||
docker run \
|
||||
--rm \
|
||||
--volume=`pwd`:/go/src/github.com/coreos/etcd \
|
||||
gcr.io/etcd-development/etcd-test:go1.8.7 \
|
||||
/bin/bash -c "${TEST_OPTS} ./test 2>&1 | tee test-${TEST_SUFFIX}.log"
|
||||
|
||||
! egrep "(--- FAIL:|panic: test timed out|appears to have leaked)" -B50 -A10 test-${TEST_SUFFIX}.log
|
||||
echo "Running tests with" ${TEST_OPTS}
|
||||
if [ "${TEST_OPTS}" == "PASSES='build grpcproxy'" ]; then
|
||||
echo "Skip proxy tests for this branch!"
|
||||
exit 0
|
||||
elif [ "${TEST_OPTS}" == "coverage" ]; then
|
||||
echo "Skip coverage tests for this branch!"
|
||||
exit 0
|
||||
else
|
||||
sudo HOST_TMP_DIR=/tmp TEST_OPTS="${TEST_OPTS}" make docker-test
|
||||
fi
|
||||
|
@@ -26,7 +26,7 @@ import (
|
||||
var (
|
||||
// MinClusterVersion is the min cluster version this etcd binary is compatible with.
|
||||
MinClusterVersion = "3.0.0"
|
||||
Version = "3.2.19"
|
||||
Version = "3.2.23"
|
||||
APIVersion = "unknown"
|
||||
|
||||
// Git SHA Value will be set during build
|
||||
|
Reference in New Issue
Block a user