diff --git a/weed/command/scaffold/filer.toml b/weed/command/scaffold/filer.toml index 8b9552e052..ef71215c2f 100644 --- a/weed/command/scaffold/filer.toml +++ b/weed/command/scaffold/filer.toml @@ -365,7 +365,15 @@ servers = "localhost:2379" username = "" password = "" key_prefix = "seaweedfs." +# how long connecting to etcd may take timeout = "3s" +# how long an etcd read or delete may take, unless the filer already set a deadline for it; +# "0s" sets none. Writes have no such deadline, since a timed-out write may still be saved. +request_timeout = "3s" +# drop the connection to an etcd member that leaves sent data unacknowledged (Linux only), +# or a keepalive ping unanswered, for this long, so requests go to the other members. +# The ping goes out after 10s without traffic from the member. "0s" turns keepalive off. +keepalive_timeout = "3s" # Set the CA certificate path tls_ca_file="" # Set the client certificate path diff --git a/weed/filer/etcd/etcd_store.go b/weed/filer/etcd/etcd_store.go index 6cf7dd880d..86656e8b9c 100644 --- a/weed/filer/etcd/etcd_store.go +++ b/weed/filer/etcd/etcd_store.go @@ -19,6 +19,11 @@ import ( const ( DIR_FILE_SEPARATOR = byte(0x00) + + // keepaliveTime is how long a connection with calls in flight may go + // without receiving anything before the client pings the member. It is + // gRPC's minimum: grpc.WithKeepaliveParams raises anything lower to 10s. + keepaliveTime = 10 * time.Second ) func init() { @@ -26,9 +31,11 @@ func init() { } type EtcdStore struct { - client *clientv3.Client - etcdKeyPrefix string - timeout time.Duration + client *clientv3.Client + etcdKeyPrefix string + timeout time.Duration + requestTimeout time.Duration + keepaliveTimeout time.Duration } func (store *EtcdStore) GetName() string { @@ -38,6 +45,8 @@ func (store *EtcdStore) GetName() string { func (store *EtcdStore) Initialize(configuration weed_util.Configuration, prefix string) error { configuration.SetDefault(prefix+"servers", "localhost:2379") configuration.SetDefault(prefix+"timeout", "3s") + configuration.SetDefault(prefix+"request_timeout", "3s") + configuration.SetDefault(prefix+"keepalive_timeout", "3s") servers := configuration.GetString(prefix + "servers") @@ -52,6 +61,13 @@ func (store *EtcdStore) Initialize(configuration weed_util.Configuration, prefix } store.timeout = timeout + if store.requestTimeout, err = time.ParseDuration(configuration.GetString(prefix + "request_timeout")); err != nil { + return fmt.Errorf("parse etcd store request_timeout: %w", err) + } + if store.keepaliveTimeout, err = time.ParseDuration(configuration.GetString(prefix + "keepalive_timeout")); err != nil { + return fmt.Errorf("parse etcd store keepalive_timeout: %w", err) + } + certFile := configuration.GetString(prefix + "tls_client_crt_file") keyFile := configuration.GetString(prefix + "tls_client_key_file") caFile := configuration.GetString(prefix + "tls_ca_file") @@ -76,13 +92,23 @@ func (store *EtcdStore) Initialize(configuration weed_util.Configuration, prefix func (store *EtcdStore) initialize(servers, username, password string, timeout time.Duration, tlsConfig *tls.Config) error { glog.Infof("filer store etcd: %s", servers) - client, err := clientv3.New(clientv3.Config{ + config := clientv3.Config{ Endpoints: strings.Split(servers, ","), Username: username, Password: password, DialTimeout: timeout, TLS: tlsConfig, - }) + } + // Keepalive closes the connection to a member that stops answering, so the + // balancer stops sending requests to it. gRPC also sets TCP_USER_TIMEOUT to + // keepaliveTimeout on Linux: a connection whose sent data stays + // unacknowledged that long is closed by the kernel, which catches a dead + // node or a network partition without waiting for keepaliveTime. + if store.keepaliveTimeout > 0 { + config.DialKeepAliveTime = keepaliveTime + config.DialKeepAliveTimeout = store.keepaliveTimeout + } + client, err := clientv3.New(config) if err != nil { return fmt.Errorf("connect to etcd %s: %s", servers, err) } @@ -112,6 +138,17 @@ func (store *EtcdStore) RollbackTransaction(ctx context.Context) error { return nil } +// withTimeout bounds one etcd call by the store's request timeout, unless the +// caller has set a deadline of its own. A call without a deadline that is balanced +// onto a member that stops answering waits until keepalive closes the +// connection, or forever if keepalive is off. +func (store *EtcdStore) withTimeout(ctx context.Context) (context.Context, context.CancelFunc) { + if _, hasDeadline := ctx.Deadline(); hasDeadline || store.requestTimeout <= 0 { + return ctx, func() {} + } + return context.WithTimeout(ctx, store.requestTimeout) +} + func (store *EtcdStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) { key := genKey(entry.DirAndName()) @@ -124,6 +161,11 @@ func (store *EtcdStore) InsertEntry(ctx context.Context, entry *filer.Entry) (er meta = weed_util.MaybeGzipData(meta) } + // No request deadline here: a Put that times out on the client may still + // have been committed, and callers such as the filer's upload handler treat + // the error as a failed write and delete the chunks the saved entry points + // at. etcd's own request timeout still ends a Put that a live member cannot + // commit, and keepalive ends one on a member that goes silent. if _, err := store.client.Put(ctx, store.etcdKeyPrefix+string(key), string(meta)); err != nil { return fmt.Errorf("persisting %s : %v", entry.FullPath, err) } @@ -138,7 +180,9 @@ func (store *EtcdStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (er func (store *EtcdStore) FindEntry(ctx context.Context, fullpath weed_util.FullPath) (entry *filer.Entry, err error) { key := genKey(fullpath.DirAndName()) - resp, err := store.client.Get(ctx, store.etcdKeyPrefix+string(key), atScanRevision(ctx)...) + opCtx, cancel := store.withTimeout(ctx) + defer cancel() + resp, err := store.client.Get(opCtx, store.etcdKeyPrefix+string(key), atScanRevision(ctx)...) if err != nil { return nil, fmt.Errorf("get %s : %v", fullpath, err) } @@ -161,13 +205,19 @@ func (store *EtcdStore) FindEntry(ctx context.Context, fullpath weed_util.FullPa func (store *EtcdStore) DeleteEntry(ctx context.Context, fullpath weed_util.FullPath) (err error) { key := genKey(fullpath.DirAndName()) - if _, err := store.client.Delete(ctx, store.etcdKeyPrefix+string(key)); err != nil { + opCtx, cancel := store.withTimeout(ctx) + defer cancel() + if _, err := store.client.Delete(opCtx, store.etcdKeyPrefix+string(key)); err != nil { return fmt.Errorf("delete %s : %v", fullpath, err) } return nil } +// DeleteFolderChildren removes every entry under fullpath in one DeleteRange. +// It runs without the store's timeout: the range grows with the directory +// (about 3.5 s for 600,000 entries), and keepalive still fails the call if the +// member stops answering. func (store *EtcdStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) { directoryPrefix := genDirectoryKeyPrefix(fullpath, "") @@ -185,7 +235,9 @@ func (store *EtcdStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPat lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName) } - resp, err := store.client.Get(ctx, store.etcdKeyPrefix+string(lastFileStart), + opCtx, cancel := store.withTimeout(ctx) + defer cancel() + resp, err := store.client.Get(opCtx, store.etcdKeyPrefix+string(lastFileStart), clientv3.WithRange(clientv3.GetPrefixRangeEnd(store.etcdKeyPrefix+string(directoryPrefix))), clientv3.WithLimit(limit+1)) if err != nil { diff --git a/weed/filer/etcd/etcd_store_kv.go b/weed/filer/etcd/etcd_store_kv.go index 15bcb131c2..f762ae05c5 100644 --- a/weed/filer/etcd/etcd_store_kv.go +++ b/weed/filer/etcd/etcd_store_kv.go @@ -7,6 +7,9 @@ import ( "github.com/seaweedfs/seaweedfs/weed/filer" ) +// KvPut has no request deadline, like InsertEntry: the filer stores a hard-linked +// file's chunk list here, so a timed-out Put that was saved would leave it +// pointing at chunks the upload handler deletes. func (store *EtcdStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { _, err = store.client.Put(ctx, store.etcdKeyPrefix+string(key), string(value)) @@ -19,8 +22,10 @@ func (store *EtcdStore) KvPut(ctx context.Context, key []byte, value []byte) (er } func (store *EtcdStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { + opCtx, cancel := store.withTimeout(ctx) + defer cancel() - resp, err := store.client.Get(ctx, store.etcdKeyPrefix+string(key), atScanRevision(ctx)...) + resp, err := store.client.Get(opCtx, store.etcdKeyPrefix+string(key), atScanRevision(ctx)...) if err != nil { return nil, fmt.Errorf("kv get: %w", err) @@ -34,8 +39,10 @@ func (store *EtcdStore) KvGet(ctx context.Context, key []byte) (value []byte, er } func (store *EtcdStore) KvDelete(ctx context.Context, key []byte) (err error) { + opCtx, cancel := store.withTimeout(ctx) + defer cancel() - _, err = store.client.Delete(ctx, store.etcdKeyPrefix+string(key)) + _, err = store.client.Delete(opCtx, store.etcdKeyPrefix+string(key)) if err != nil { return fmt.Errorf("kv delete: %w", err) diff --git a/weed/filer/etcd/etcd_store_test.go b/weed/filer/etcd/etcd_store_test.go index 127e10cb01..7dee3bfc0c 100644 --- a/weed/filer/etcd/etcd_store_test.go +++ b/weed/filer/etcd/etcd_store_test.go @@ -1,9 +1,13 @@ package etcd import ( + "context" "testing" + "time" + "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/filer/store_test" + clientv3 "go.etcd.io/etcd/client/v3" ) func TestStore(t *testing.T) { @@ -15,3 +19,65 @@ func TestStore(t *testing.T) { store_test.TestFilerStore(t, store) } } + +func TestWithTimeout(t *testing.T) { + store := &EtcdStore{timeout: time.Minute, requestTimeout: 3 * time.Second} + + ctx, cancel := store.withTimeout(context.Background()) + defer cancel() + deadline, ok := ctx.Deadline() + if !ok || time.Until(deadline) > store.requestTimeout { + t.Fatalf("no deadline within the store's request timeout: %v %v", deadline, ok) + } + + callerCtx, callerCancel := context.WithTimeout(context.Background(), time.Minute) + defer callerCancel() + ctx, cancel = store.withTimeout(callerCtx) + defer cancel() + if ctx != callerCtx { + t.Fatal("the caller's deadline was replaced") + } + + store.requestTimeout = 0 + ctx, cancel = store.withTimeout(context.Background()) + defer cancel() + if _, ok := ctx.Deadline(); ok { + t.Fatal(`request_timeout "0s" set a deadline`) + } +} + +type putDeadlineKV struct { + clientv3.KV + deadlines []bool +} + +func (kv *putDeadlineKV) Put(ctx context.Context, key, val string, opts ...clientv3.OpOption) (*clientv3.PutResponse, error) { + _, ok := ctx.Deadline() + kv.deadlines = append(kv.deadlines, ok) + return &clientv3.PutResponse{}, nil +} + +func TestWritesHaveNoRequestDeadline(t *testing.T) { + kv := &putDeadlineKV{} + store := &EtcdStore{client: &clientv3.Client{KV: kv}, requestTimeout: 3 * time.Second} + ctx := context.Background() + entry := &filer.Entry{FullPath: "/dir/file"} + + if err := store.InsertEntry(ctx, entry); err != nil { + t.Fatal(err) + } + if err := store.UpdateEntry(ctx, entry); err != nil { + t.Fatal(err) + } + if err := store.KvPut(ctx, []byte("hard link id"), []byte("chunks")); err != nil { + t.Fatal(err) + } + if len(kv.deadlines) != 3 { + t.Fatalf("got %d puts, want 3", len(kv.deadlines)) + } + for i, hasDeadline := range kv.deadlines { + if hasDeadline { + t.Errorf("put %d has a request deadline", i) + } + } +}