filer/etcd: client keepalive and per-request deadlines (#12621)

* filer/etcd: client keepalive and per-request deadlines An etcd member that goes silent with its connections open (a dead or hung node, a network partition, a frozen process) keeps the store's requests balanced onto it waiting forever, because the client has no keepalive and the calls have no deadline. Writes through the filer stop although the other members keep quorum. keepalive_timeout (default 3s, "0s" turns keepalive off) enables gRPC keepalive with the ping interval fixed at gRPC's 10s minimum. gRPC also sets TCP_USER_TIMEOUT to it on Linux, so a connection whose sent data goes unacknowledged that long is closed by the kernel. request_timeout (default 3s) bounds every Get, Put and Delete unless the caller set a deadline; timeout keeps its meaning as the dial timeout. DeleteFolderChildren gets no request deadline: its DeleteRange grows with the directory, and keepalive still ends it if the member goes silent. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Update weed/filer/etcd/etcd_store.go Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com> * Update weed/filer/etcd/etcd_store_test.go Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com> * Update weed/command/scaffold/filer.toml Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com> * filer/etcd: no request deadline on entry writes A Put whose client deadline expires may still have been committed. The filer's upload handler treats the error as a failed write and deletes the uploaded chunks, which the saved entry may then point at. InsertEntry and UpdateEntry therefore keep the caller's context: etcd's own request timeout ends a Put a live member cannot commit, and keepalive ends one on a member that goes silent. Reads and deletes keep request_timeout; a timed-out store delete returns before the filer deletes any chunks. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * filer/etcd: no request deadline on KvPut The filer wrapper stores a hard-linked file's attributes and chunk list through KvPut. An HTTP append to a hard-linked file whose KvPut times out after etcd saved it returns an error, and the upload handler deletes the appended chunks the saved hard link now points at. KvPut keeps the caller's context like InsertEntry and UpdateEntry. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
2026-11-15 08:18:42 +01:00 · 2026-11-13 12:28:48 -08:00
1 parent 911bfe4014
commit 5a54009439
@@ -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
@@ -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 {
@@ -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)
@@ -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)
+		}
+	}
+}