diff --git a/stress b/stress new file mode 100755 index 000000000..c2a6445fa Binary files /dev/null and b/stress differ diff --git a/unmaintained/stress/ec_stress_runner.go b/unmaintained/stress/ec_stress_runner.go index 3acd36c0d..239f15988 100644 --- a/unmaintained/stress/ec_stress_runner.go +++ b/unmaintained/stress/ec_stress_runner.go @@ -301,7 +301,7 @@ func (r *runner) fetchVolumeList(ctx context.Context) (*master_pb.VolumeListResp } var response *master_pb.VolumeListResponse - err := pb.WithMasterClient(false, pb.ServerAddress(masterAddress), r.grpcDialOption, false, func(client master_pb.SeaweedClient) error { + err := pb.WithMasterClient(context.Background(), false, pb.ServerAddress(masterAddress), r.grpcDialOption, false, func(client master_pb.SeaweedClient) error { callCtx, cancel := context.WithTimeout(ctx, r.cfg.RequestTimeout) defer cancel() diff --git a/weed/cluster/master_client.go b/weed/cluster/master_client.go index ce41b0515..ba0ed4f3d 100644 --- a/weed/cluster/master_client.go +++ b/weed/cluster/master_client.go @@ -12,7 +12,7 @@ import ( func ListExistingPeerUpdates(master pb.ServerAddress, grpcDialOption grpc.DialOption, filerGroup string, clientType string) (existingNodes []*master_pb.ClusterNodeUpdate) { - if grpcErr := pb.WithMasterClient(false, master, grpcDialOption, false, func(client master_pb.SeaweedClient) error { + if grpcErr := pb.WithMasterClient(context.Background(), false, master, grpcDialOption, false, func(client master_pb.SeaweedClient) error { resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{ ClientType: clientType, FilerGroup: filerGroup, diff --git a/weed/command/filer_copy.go b/weed/command/filer_copy.go index e849c8e07..f4a712f8f 100644 --- a/weed/command/filer_copy.go +++ b/weed/command/filer_copy.go @@ -596,7 +596,7 @@ var _ = filer_pb.FilerClient(&FileCopyWorker{}) func (worker *FileCopyWorker) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) (err error) { filerGrpcAddress := worker.filerAddress.ToGrpcAddress() - err = pb.WithGrpcClient(streamingMode, worker.signature, func(grpcConnection *grpc.ClientConn) error { + err = pb.WithGrpcClient(context.Background(), streamingMode, worker.signature, func(grpcConnection *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(grpcConnection) return fn(client) }, filerGrpcAddress, false, worker.options.grpcDialOption) diff --git a/weed/command/filer_sync_verify.go b/weed/command/filer_sync_verify.go index b60e5f60a..366f9e4fe 100644 --- a/weed/command/filer_sync_verify.go +++ b/weed/command/filer_sync_verify.go @@ -177,7 +177,7 @@ type simpleFilerClient struct { } func (c *simpleFilerClient) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error { - return pb.WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { + return pb.WithGrpcClient(context.Background(), streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(grpcConnection) return fn(client) }, c.grpcAddress.ToGrpcAddress(), false, c.grpcDialOption) diff --git a/weed/command/upload.go b/weed/command/upload.go index 9f9ac1107..313d7b4a6 100644 --- a/weed/command/upload.go +++ b/weed/command/upload.go @@ -145,7 +145,7 @@ func runUpload(cmd *Command, args []string) bool { } func readMasterConfiguration(grpcDialOption grpc.DialOption, masterAddress pb.ServerAddress) (replication string, err error) { - err = pb.WithMasterClient(false, masterAddress, grpcDialOption, false, func(client master_pb.SeaweedClient) error { + err = pb.WithMasterClient(context.Background(), false, masterAddress, grpcDialOption, false, func(client master_pb.SeaweedClient) error { resp, err := client.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{}) if err != nil { return fmt.Errorf("get master %s configuration: %v", masterAddress, err) diff --git a/weed/credential/grpc/grpc_store.go b/weed/credential/grpc/grpc_store.go index 9b59ba0a4..944a09c13 100644 --- a/weed/credential/grpc/grpc_store.go +++ b/weed/credential/grpc/grpc_store.go @@ -98,7 +98,7 @@ func (store *IamGrpcStore) withIamClient(ctx context.Context, fn func(ctx contex } } - return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error { + return pb.WithGrpcClient(context.Background(), false, 0, func(conn *grpc.ClientConn) error { client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn) return fn(ctx, client) }, filerAddress.ToGrpcAddress(), false, dialOption) diff --git a/weed/credential/propagating_store.go b/weed/credential/propagating_store.go index ca5bbd577..050aff28e 100644 --- a/weed/credential/propagating_store.go +++ b/weed/credential/propagating_store.go @@ -86,7 +86,7 @@ func (s *PropagatingCredentialStore) propagateChange(ctx context.Context, fn fun wg.Add(1) go func(server string) { defer wg.Done() - err := pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error { + err := pb.WithGrpcClient(context.Background(), false, 0, func(conn *grpc.ClientConn) error { glog.V(4).Infof("IAM: successfully connected to S3 server %s for propagation", server) client := s3_pb.NewSeaweedS3IamCacheClient(conn) return fn(propagateCtx, client) diff --git a/weed/mount/wfs_filer_client.go b/weed/mount/wfs_filer_client.go index d12fca76e..d3e8da546 100644 --- a/weed/mount/wfs_filer_client.go +++ b/weed/mount/wfs_filer_client.go @@ -1,6 +1,7 @@ package mount import ( + "context" "sync/atomic" "google.golang.org/grpc" @@ -21,7 +22,7 @@ func (wfs *WFS) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFile for x := 0; x < n; x++ { filerGrpcAddress := wfs.option.FilerAddresses[i].ToGrpcAddress() - err = pb.WithGrpcClient(streamingMode, wfs.signature, func(grpcConnection *grpc.ClientConn) error { + err = pb.WithGrpcClient(context.Background(), streamingMode, wfs.signature, func(grpcConnection *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(grpcConnection) return fn(client) }, filerGrpcAddress, false, wfs.option.GrpcDialOption) diff --git a/weed/operation/assign_file_id.go b/weed/operation/assign_file_id.go index 783d1a204..70c54b9ed 100644 --- a/weed/operation/assign_file_id.go +++ b/weed/operation/assign_file_id.go @@ -91,7 +91,10 @@ func Assign(ctx context.Context, masterFn GetMasterFn, grpcDialOption grpc.DialO // Per-attempt timeout to prevent a single slow RPC from consuming the entire retry budget attemptCtx, attemptCancel := context.WithTimeout(deadlineCtx, 10*time.Second) defer attemptCancel() - return WithMasterServerClient(false, masterFn(attemptCtx), grpcDialOption, func(masterClient master_pb.SeaweedClient) error { + // Pass attemptCtx so its expiry is not mistaken for a dead shared + // connection: invalidating it would cancel every other in-flight + // assign with "the client connection is closing". + return WithMasterServerClient(attemptCtx, false, masterFn(attemptCtx), grpcDialOption, func(masterClient master_pb.SeaweedClient) error { req := &master_pb.AssignRequest{ Count: request.Count, Replication: request.Replication, @@ -151,7 +154,7 @@ func Assign(ctx context.Context, masterFn GetMasterFn, grpcDialOption grpc.DialO func LookupJwt(master pb.ServerAddress, grpcDialOption grpc.DialOption, fileId string) (token security.EncodedJwt) { - WithMasterServerClient(false, master, grpcDialOption, func(masterClient master_pb.SeaweedClient) error { + WithMasterServerClient(context.Background(), false, master, grpcDialOption, func(masterClient master_pb.SeaweedClient) error { resp, grpcErr := masterClient.LookupVolume(context.Background(), &master_pb.LookupVolumeRequest{ VolumeOrFileIds: []string{fileId}, diff --git a/weed/operation/grpc_client.go b/weed/operation/grpc_client.go index ecd8117ee..9e15ba115 100644 --- a/weed/operation/grpc_client.go +++ b/weed/operation/grpc_client.go @@ -1,6 +1,8 @@ package operation import ( + "context" + "google.golang.org/grpc" "github.com/seaweedfs/seaweedfs/weed/pb" @@ -10,18 +12,17 @@ import ( func WithVolumeServerClient(streamingMode bool, volumeServer pb.ServerAddress, grpcDialOption grpc.DialOption, fn func(volume_server_pb.VolumeServerClient) error) error { - return pb.WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { + return pb.WithGrpcClient(context.Background(), streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { client := volume_server_pb.NewVolumeServerClient(grpcConnection) return fn(client) }, volumeServer.ToGrpcAddress(), false, grpcDialOption) } -func WithMasterServerClient(streamingMode bool, masterServer pb.ServerAddress, grpcDialOption grpc.DialOption, fn func(masterClient master_pb.SeaweedClient) error) error { - - return pb.WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { - client := master_pb.NewSeaweedClient(grpcConnection) - return fn(client) - }, masterServer.ToGrpcAddress(), false, grpcDialOption) - +// WithMasterServerClient threads the caller's per-request context into the +// connection-invalidation decision, so a Canceled/DeadlineExceeded from the +// caller's own timeout does not invalidate the shared cached master connection. +// Pass context.Background() when there is no per-request deadline to honor. +func WithMasterServerClient(ctx context.Context, streamingMode bool, masterServer pb.ServerAddress, grpcDialOption grpc.DialOption, fn func(masterClient master_pb.SeaweedClient) error) error { + return pb.WithMasterClient(ctx, streamingMode, masterServer, grpcDialOption, false, fn) } diff --git a/weed/operation/lookup.go b/weed/operation/lookup.go index aedb8f8ee..0580c9f95 100644 --- a/weed/operation/lookup.go +++ b/weed/operation/lookup.go @@ -82,7 +82,7 @@ func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids //only query unknown_vids - err := WithMasterServerClient(false, masterFn(context.Background()), grpcDialOption, func(masterClient master_pb.SeaweedClient) error { + err := WithMasterServerClient(context.Background(), false, masterFn(context.Background()), grpcDialOption, func(masterClient master_pb.SeaweedClient) error { req := &master_pb.LookupVolumeRequest{ VolumeOrFileIds: unknown_vids, diff --git a/weed/pb/grpc_client_server.go b/weed/pb/grpc_client_server.go index b3aff37a9..36db5579a 100644 --- a/weed/pb/grpc_client_server.go +++ b/weed/pb/grpc_client_server.go @@ -311,7 +311,9 @@ func InvalidateGrpcConnection(address string) { // grpcMarshalErrorPrefix is the library-owned prefix gRPC prepends to every // client-side proto marshal failure; see grpc-go rpc_util.go encode(): -// status.Errorf(codes.Internal, "grpc: error while marshaling: %v", ...) +// +// status.Errorf(codes.Internal, "grpc: error while marshaling: %v", ...) +// // The "grpc:" token is reserved for gRPC internal diagnostics and will not // collide with user-produced Internal statuses. const grpcMarshalErrorPrefix = "grpc: error while marshaling" @@ -352,8 +354,11 @@ func isClientSideMarshalError(err error) bool { return s.Code() == codes.Internal && strings.HasPrefix(s.Message(), grpcMarshalErrorPrefix) } -// shouldInvalidateConnection checks if an error indicates the cached connection should be invalidated -func shouldInvalidateConnection(err error) bool { +// shouldInvalidateConnection checks if an error indicates the cached connection +// should be invalidated. ctx is the caller's per-request context (nil is treated +// as a live context); it disambiguates a genuinely broken channel from the +// caller cancelling or timing out its own request. +func shouldInvalidateConnection(ctx context.Context, err error) bool { if err == nil { return false } @@ -369,8 +374,19 @@ func shouldInvalidateConnection(err error) bool { if s, ok := status.FromError(err); ok { code := s.Code() switch code { - case codes.Unavailable, codes.Canceled, codes.DeadlineExceeded, codes.Aborted, codes.Internal: + case codes.Unavailable, codes.Aborted, codes.Internal: return true + case codes.Canceled, codes.DeadlineExceeded: + // Ambiguous: this fires both when a stale cached channel rejects + // RPCs (e.g. a peer restart behind a k8s Service VIP), where we must + // invalidate, and when the caller's own context expired, where the + // shared channel is fine. Tearing the channel down for the latter + // cancels every other in-flight RPC on it with "the client + // connection is closing", a cascade that turns one slow request into + // a flood of failures across all concurrent callers. Only invalidate + // while the caller's context is still live, isolating the genuine + // stale-channel signal. + return ctx == nil || ctx.Err() == nil } } @@ -387,7 +403,11 @@ func shouldInvalidateConnection(err error) bool { } // WithGrpcClient In streamingMode, always use a fresh connection. Otherwise, try to reuse an existing connection. -func WithGrpcClient(streamingMode bool, signature int32, fn func(*grpc.ClientConn) error, address string, waitForReady bool, opts ...grpc.DialOption) error { +// ctx is the caller's per-request context: a Canceled/DeadlineExceeded that fires +// because ctx itself expired will not tear down the shared cached ClientConn out +// from under other concurrent callers. Pass context.Background() when there is no +// per-request deadline to honor. +func WithGrpcClient(ctx context.Context, streamingMode bool, signature int32, fn func(*grpc.ClientConn) error, address string, waitForReady bool, opts ...grpc.DialOption) error { if !streamingMode { vgc, err := getOrCreateConnection(address, waitForReady, opts...) @@ -395,7 +415,7 @@ func WithGrpcClient(streamingMode bool, signature int32, fn func(*grpc.ClientCon return fmt.Errorf("getOrCreateConnection %s: %v", address, err) } executionErr := fn(vgc.ClientConn) - if executionErr != nil && shouldInvalidateConnection(executionErr) { + if executionErr != nil && shouldInvalidateConnection(ctx, executionErr) { grpcClientsLock.Lock() t, ok := grpcClients[address] shouldClose := ok && t.version == vgc.version @@ -412,12 +432,12 @@ func WithGrpcClient(streamingMode bool, signature int32, fn func(*grpc.ClientCon } // Streaming mode: dedicate a fresh ClientConn to this call. - ctx := context.Background() + dialCtx := context.Background() if signature != 0 { // Optimize: Use AppendToOutgoingContext instead of creating new map - ctx = metadata.AppendToOutgoingContext(ctx, "sw-client-id", fmt.Sprintf("%d", signature)) + dialCtx = metadata.AppendToOutgoingContext(dialCtx, "sw-client-id", fmt.Sprintf("%d", signature)) } - grpcConnection, err := GrpcDial(ctx, address, waitForReady, opts...) + grpcConnection, err := GrpcDial(dialCtx, address, waitForReady, opts...) if err != nil { return fmt.Errorf("fail to dial %s: %v", address, err) } @@ -494,8 +514,12 @@ func GrpcAddressToServerAddress(grpcAddress string) (serverAddress string) { return util.JoinHostPort(host, port) } -func WithMasterClient(streamingMode bool, master ServerAddress, grpcDialOption grpc.DialOption, waitForReady bool, fn func(client master_pb.SeaweedClient) error) error { - return WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { +// WithMasterClient threads the caller's per-request context into the +// connection-invalidation decision, so a Canceled/DeadlineExceeded from the +// caller's own timeout does not invalidate the shared cached master connection. +// Pass context.Background() when there is no per-request deadline to honor. +func WithMasterClient(ctx context.Context, streamingMode bool, master ServerAddress, grpcDialOption grpc.DialOption, waitForReady bool, fn func(client master_pb.SeaweedClient) error) error { + return WithGrpcClient(ctx, streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { client := master_pb.NewSeaweedClient(grpcConnection) return fn(client) }, master.ToGrpcAddress(), waitForReady, grpcDialOption) @@ -503,7 +527,7 @@ func WithMasterClient(streamingMode bool, master ServerAddress, grpcDialOption g } func WithVolumeServerClient(streamingMode bool, volumeServer ServerAddress, grpcDialOption grpc.DialOption, fn func(client volume_server_pb.VolumeServerClient) error) error { - return WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { + return WithGrpcClient(context.Background(), streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { client := volume_server_pb.NewVolumeServerClient(grpcConnection) return fn(client) }, volumeServer.ToGrpcAddress(), false, grpcDialOption) @@ -513,7 +537,7 @@ func WithVolumeServerClient(streamingMode bool, volumeServer ServerAddress, grpc func WithOneOfGrpcMasterClients(streamingMode bool, masterGrpcAddresses map[string]ServerAddress, grpcDialOption grpc.DialOption, fn func(client master_pb.SeaweedClient) error) (err error) { for _, masterGrpcAddress := range masterGrpcAddresses { - err = WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { + err = WithGrpcClient(context.Background(), streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { client := master_pb.NewSeaweedClient(grpcConnection) return fn(client) }, masterGrpcAddress.ToGrpcAddress(), false, grpcDialOption) @@ -527,7 +551,7 @@ func WithOneOfGrpcMasterClients(streamingMode bool, masterGrpcAddresses map[stri func WithBrokerGrpcClient(streamingMode bool, brokerGrpcAddress string, grpcDialOption grpc.DialOption, fn func(client mq_pb.SeaweedMessagingClient) error) error { - return WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { + return WithGrpcClient(context.Background(), streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { client := mq_pb.NewSeaweedMessagingClient(grpcConnection) return fn(client) }, brokerGrpcAddress, false, grpcDialOption) @@ -542,7 +566,7 @@ func WithFilerClient(streamingMode bool, signature int32, filer ServerAddress, g func WithGrpcFilerClient(streamingMode bool, signature int32, filerAddress ServerAddress, grpcDialOption grpc.DialOption, fn func(client filer_pb.SeaweedFilerClient) error) error { - return WithGrpcClient(streamingMode, signature, func(grpcConnection *grpc.ClientConn) error { + return WithGrpcClient(context.Background(), streamingMode, signature, func(grpcConnection *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(grpcConnection) return fn(client) }, filerAddress.ToGrpcAddress(), false, grpcDialOption) @@ -552,7 +576,7 @@ func WithGrpcFilerClient(streamingMode bool, signature int32, filerAddress Serve func WithOneOfGrpcFilerClients(streamingMode bool, filerAddresses []ServerAddress, grpcDialOption grpc.DialOption, fn func(client filer_pb.SeaweedFilerClient) error) (err error) { for _, filerAddress := range filerAddresses { - err = WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { + err = WithGrpcClient(context.Background(), streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(grpcConnection) return fn(client) }, filerAddress.ToGrpcAddress(), false, grpcDialOption) diff --git a/weed/pb/grpc_client_server_test.go b/weed/pb/grpc_client_server_test.go index bdc3cc48b..e38514111 100644 --- a/weed/pb/grpc_client_server_test.go +++ b/weed/pb/grpc_client_server_test.go @@ -1,6 +1,7 @@ package pb import ( + "context" "fmt" "runtime" "testing" @@ -19,24 +20,59 @@ func TestShouldInvalidateConnection_MarshalErrorIsPerRequest(t *testing.T) { // outgoing request contains invalid UTF-8 bytes. marshalErr := status.Error(codes.Internal, "grpc: error while marshaling: string field contains invalid UTF-8") - if shouldInvalidateConnection(marshalErr) { + if shouldInvalidateConnection(context.Background(), marshalErr) { t.Fatalf("client-side marshal error must not invalidate the shared connection") } // Same error wrapped with fmt.Errorf (common when callers add context). wrapped := fmt.Errorf("upload data: %w", marshalErr) - if shouldInvalidateConnection(wrapped) { + if shouldInvalidateConnection(context.Background(), wrapped) { t.Fatalf("wrapped marshal error must not invalidate the shared connection") } } +// TestShouldInvalidateConnection_CallerContextExpiryIsPerRequest ensures that a +// Canceled/DeadlineExceeded caused by the caller's own context expiring does NOT +// tear down the shared cached ClientConn. Doing so would cancel every other +// in-flight RPC on it with "the client connection is closing" — the cascade +// that turned one slow chunk assign into a flood of failures during a +// high-concurrency upload (seaweedfs#9765). +func TestShouldInvalidateConnection_CallerContextExpiryIsPerRequest(t *testing.T) { + expired, cancel := context.WithCancel(context.Background()) + cancel() + for _, code := range []codes.Code{codes.Canceled, codes.DeadlineExceeded} { + err := status.Error(code, "context expired") + if shouldInvalidateConnection(expired, err) { + t.Fatalf("%v with an expired caller context must not invalidate the shared connection", code) + } + } +} + +// TestShouldInvalidateConnection_StaleChannelStillInvalidates ensures the +// carve-out above is gated on the caller's context: a Canceled/DeadlineExceeded +// while the context is still live is the genuine stale-channel signal (e.g. a +// peer restart behind a k8s Service VIP) and must still invalidate so the next +// attempt dials fresh. +func TestShouldInvalidateConnection_StaleChannelStillInvalidates(t *testing.T) { + for _, code := range []codes.Code{codes.Canceled, codes.DeadlineExceeded} { + err := status.Error(code, "the client connection is closing") + if !shouldInvalidateConnection(context.Background(), err) { + t.Fatalf("%v with a live caller context must still invalidate the connection", code) + } + // nil context is treated as live for the context-free WithGrpcClient path. + if !shouldInvalidateConnection(nil, err) { + t.Fatalf("%v with a nil caller context must still invalidate the connection", code) + } + } +} + // TestShouldInvalidateConnection_GenuineInternalStillInvalidates ensures the // marshal-error carve-out does not swallow real server-side Internal errors, // which previously caused — and should continue to cause — connection // invalidation. func TestShouldInvalidateConnection_GenuineInternalStillInvalidates(t *testing.T) { serverInternal := status.Error(codes.Internal, "stream terminated by RST_STREAM with code 2") - if !shouldInvalidateConnection(serverInternal) { + if !shouldInvalidateConnection(context.Background(), serverInternal) { t.Fatalf("genuine server-side Internal must still invalidate the connection") } } @@ -50,7 +86,7 @@ func TestShouldInvalidateConnection_TransportErrorsStillInvalidate(t *testing.T) "dial tcp: connection refused", "read: connection reset by peer", } { - if !shouldInvalidateConnection(fmt.Errorf("%s", msg)) { + if !shouldInvalidateConnection(context.Background(), fmt.Errorf("%s", msg)) { t.Fatalf("transport error %q must still invalidate", msg) } } diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go index b1fcf3525..550c59d6c 100644 --- a/weed/replication/sink/filersink/fetch_write.go +++ b/weed/replication/sink/filersink/fetch_write.go @@ -509,7 +509,7 @@ var _ = filer_pb.FilerClient(&FilerSink{}) func (fs *FilerSink) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error { - return pb.WithGrpcClient(streamingMode, fs.signature, func(grpcConnection *grpc.ClientConn) error { + return pb.WithGrpcClient(context.Background(), streamingMode, fs.signature, func(grpcConnection *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(grpcConnection) return fn(client) }, fs.grpcAddress, false, fs.grpcDialOption) diff --git a/weed/replication/source/filer_source.go b/weed/replication/source/filer_source.go index ad8683488..babbb6c71 100644 --- a/weed/replication/source/filer_source.go +++ b/weed/replication/source/filer_source.go @@ -149,7 +149,7 @@ var _ = filer_pb.FilerClient(&FilerSource{}) func (fs *FilerSource) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error { - return pb.WithGrpcClient(streamingMode, fs.signature, func(grpcConnection *grpc.ClientConn) error { + return pb.WithGrpcClient(context.Background(), streamingMode, fs.signature, func(grpcConnection *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(grpcConnection) return fn(client) }, fs.grpcAddress, false, fs.grpcDialOption) diff --git a/weed/s3api/s3api_handlers.go b/weed/s3api/s3api_handlers.go index a1f46795c..9b57d5807 100644 --- a/weed/s3api/s3api_handlers.go +++ b/weed/s3api/s3api_handlers.go @@ -1,6 +1,7 @@ package s3api import ( + "context" "encoding/base64" "errors" "fmt" @@ -24,7 +25,7 @@ func (s3a *S3ApiServer) WithFilerClient(streamingMode bool, fn func(filer_pb.Sea // Fallback to direct connection if filerClient not initialized // This should only happen during initialization or testing - return pb.WithGrpcClient(streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error { + return pb.WithGrpcClient(context.Background(), streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(grpcConnection) return fn(client) }, s3a.getFilerAddress().ToGrpcAddress(), false, s3a.option.GrpcDialOption) @@ -37,7 +38,7 @@ func (s3a *S3ApiServer) withFilerClientFailover(streamingMode bool, fn func(file currentFiler := s3a.filerClient.GetCurrentFiler() // Try current filer first (fast path) - err := pb.WithGrpcClient(streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error { + err := pb.WithGrpcClient(context.Background(), streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(grpcConnection) return fn(client) }, currentFiler.ToGrpcAddress(), false, s3a.option.GrpcDialOption) @@ -70,7 +71,7 @@ func (s3a *S3ApiServer) withFilerClientFailover(streamingMode bool, fn func(file continue } - err = pb.WithGrpcClient(streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error { + err = pb.WithGrpcClient(context.Background(), streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(grpcConnection) return fn(client) }, filer.ToGrpcAddress(), false, s3a.option.GrpcDialOption) diff --git a/weed/server/filer_grpc_server_admin.go b/weed/server/filer_grpc_server_admin.go index b8282f8c9..9c9b86c31 100644 --- a/weed/server/filer_grpc_server_admin.go +++ b/weed/server/filer_grpc_server_admin.go @@ -117,7 +117,7 @@ func (fs *FilerServer) Ping(ctx context.Context, req *filer_pb.PingRequest) (res }) } if req.TargetType == cluster.MasterType { - pingErr = pb.WithMasterClient(false, pb.ServerAddress(req.Target), fs.grpcDialOption, false, func(client master_pb.SeaweedClient) error { + pingErr = pb.WithMasterClient(context.Background(), false, pb.ServerAddress(req.Target), fs.grpcDialOption, false, func(client master_pb.SeaweedClient) error { pingResp, err := client.Ping(ctx, &master_pb.PingRequest{}) if pingResp != nil { resp.RemoteTimeNs = pingResp.StartTimeNs diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index 173c771d3..697428a12 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -312,7 +312,7 @@ func (fs *FilerServer) checkWithMaster() { for !isConnected { fs.option.Masters.RefreshBySrvIfAvailable() for _, master := range fs.option.Masters.GetInstances() { - readErr := operation.WithMasterServerClient(false, master, fs.grpcDialOption, func(masterClient master_pb.SeaweedClient) error { + readErr := operation.WithMasterServerClient(context.Background(), false, master, fs.grpcDialOption, func(masterClient master_pb.SeaweedClient) error { resp, err := masterClient.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{}) if err != nil { return fmt.Errorf("get master %s configuration: %v", master, err) diff --git a/weed/server/master_grpc_server_admin.go b/weed/server/master_grpc_server_admin.go index 194ff5708..11e8e0f83 100644 --- a/weed/server/master_grpc_server_admin.go +++ b/weed/server/master_grpc_server_admin.go @@ -222,7 +222,7 @@ func (ms *MasterServer) Ping(ctx context.Context, req *master_pb.PingRequest) (r }) } if req.TargetType == cluster.MasterType { - pingErr = pb.WithMasterClient(false, pb.ServerAddress(req.Target), ms.grpcDialOption, false, func(client master_pb.SeaweedClient) error { + pingErr = pb.WithMasterClient(context.Background(), false, pb.ServerAddress(req.Target), ms.grpcDialOption, false, func(client master_pb.SeaweedClient) error { pingResp, err := client.Ping(ctx, &master_pb.PingRequest{}) if pingResp != nil { resp.RemoteTimeNs = pingResp.StartTimeNs diff --git a/weed/server/master_server.go b/weed/server/master_server.go index 14e4ed33e..45ad2988e 100644 --- a/weed/server/master_server.go +++ b/weed/server/master_server.go @@ -526,7 +526,7 @@ func (ms *MasterServer) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, startF hashicorpRaft.ServerAddress(peerAddress.ToGrpcAddress()), 0, 0) } } else { - pb.WithMasterClient(false, peerAddress, ms.grpcDialOption, true, func(client master_pb.SeaweedClient) error { + pb.WithMasterClient(context.Background(), false, peerAddress, ms.grpcDialOption, true, func(client master_pb.SeaweedClient) error { ctx, cancel := context.WithTimeout(context.TODO(), 15*time.Second) defer cancel() if _, err := client.Ping(ctx, &master_pb.PingRequest{Target: string(peerAddress), TargetType: cluster.MasterType}); err != nil { diff --git a/weed/server/volume_grpc_admin.go b/weed/server/volume_grpc_admin.go index a9b309b67..34e45e90d 100644 --- a/weed/server/volume_grpc_admin.go +++ b/weed/server/volume_grpc_admin.go @@ -280,7 +280,7 @@ func (vs *VolumeServer) makeVolumeWritable(ctx context.Context, v *storage.Volum } func (vs *VolumeServer) notifyMasterVolumeReadonly(ctx context.Context, v *storage.Volume, isReadOnly bool) error { - if grpcErr := pb.WithMasterClient(false, vs.GetMaster(ctx), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error { + if grpcErr := pb.WithMasterClient(context.Background(), false, vs.GetMaster(ctx), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error { _, err := client.VolumeMarkReadonly(context.Background(), &master_pb.VolumeMarkReadonlyRequest{ Ip: vs.store.Ip, Port: uint32(vs.store.Port), @@ -490,7 +490,7 @@ func (vs *VolumeServer) Ping(ctx context.Context, req *volume_server_pb.PingRequ }) } if req.TargetType == cluster.MasterType { - pingErr = pb.WithMasterClient(false, pb.ServerAddress(req.Target), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error { + pingErr = pb.WithMasterClient(context.Background(), false, pb.ServerAddress(req.Target), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error { pingResp, err := client.Ping(ctx, &master_pb.PingRequest{}) if pingResp != nil { resp.RemoteTimeNs = pingResp.StartTimeNs diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go index c363f823c..dc409038b 100644 --- a/weed/server/volume_grpc_client_to_master.go +++ b/weed/server/volume_grpc_client_to_master.go @@ -47,7 +47,7 @@ func (vs *VolumeServer) setCurrentMaster(master pb.ServerAddress) { func (vs *VolumeServer) checkWithMaster() (err error) { for { for _, master := range vs.SeedMasterNodes { - err = operation.WithMasterServerClient(false, master, vs.grpcDialOption, func(masterClient master_pb.SeaweedClient) error { + err = operation.WithMasterServerClient(context.Background(), false, master, vs.grpcDialOption, func(masterClient master_pb.SeaweedClient) error { resp, err := masterClient.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{}) if err != nil { return fmt.Errorf("get master %s configuration: %v", master, err) diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index 7049c031b..89202d56d 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -93,7 +93,7 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre }() var preallocateSize int64 - if grpcErr := pb.WithMasterClient(false, vs.GetMaster(context.Background()), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error { + if grpcErr := pb.WithMasterClient(context.Background(), false, vs.GetMaster(context.Background()), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error { resp, err := client.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{}) if err != nil { return fmt.Errorf("get master %s configuration: %v", vs.GetMaster(context.Background()), err) diff --git a/weed/server/webdav_server.go b/weed/server/webdav_server.go index 36f23cb44..56650ed91 100644 --- a/weed/server/webdav_server.go +++ b/weed/server/webdav_server.go @@ -145,7 +145,7 @@ var _ = filer_pb.FilerClient(&WebDavFileSystem{}) func (fs *WebDavFileSystem) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error { - return pb.WithGrpcClient(streamingMode, fs.signature, func(grpcConnection *grpc.ClientConn) error { + return pb.WithGrpcClient(context.Background(), streamingMode, fs.signature, func(grpcConnection *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(grpcConnection) return fn(client) }, fs.option.Filer.ToGrpcAddress(), false, fs.option.GrpcDialOption) diff --git a/weed/sftpd/sftp_filer.go b/weed/sftpd/sftp_filer.go index fa69e39f7..ac9f83d6c 100644 --- a/weed/sftpd/sftp_filer.go +++ b/weed/sftpd/sftp_filer.go @@ -89,7 +89,7 @@ func (fs *SftpServer) AdjustedUrl(location *filer_pb.Location) string { return l func (fs *SftpServer) GetDataCenter() string { return fs.dataCenter } func (fs *SftpServer) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error { addr := fs.filerAddr.ToGrpcAddress() - return pb.WithGrpcClient(streamingMode, util.RandomInt32(), func(conn *grpc.ClientConn) error { + return pb.WithGrpcClient(context.Background(), streamingMode, util.RandomInt32(), func(conn *grpc.ClientConn) error { return fn(filer_pb.NewSeaweedFilerClient(conn)) }, addr, false, fs.grpcDialOption) } diff --git a/weed/shell/command_cluster_check.go b/weed/shell/command_cluster_check.go index 86c58ff71..ca94ab7be 100644 --- a/weed/shell/command_cluster_check.go +++ b/weed/shell/command_cluster_check.go @@ -114,7 +114,7 @@ func (c *commandClusterCheck) Do(args []string, commandEnv *CommandEnv, writer i for _, master := range masters { for _, volumeServer := range volumeServers { fmt.Fprintf(writer, "checking master %s to volume server %s ... ", string(master), string(volumeServer)) - err := pb.WithMasterClient(false, master, commandEnv.option.GrpcDialOption, false, func(client master_pb.SeaweedClient) error { + err := pb.WithMasterClient(context.Background(), false, master, commandEnv.option.GrpcDialOption, false, func(client master_pb.SeaweedClient) error { pong, err := client.Ping(context.Background(), &master_pb.PingRequest{ Target: string(volumeServer), TargetType: cluster.VolumeServerType, @@ -137,7 +137,7 @@ func (c *commandClusterCheck) Do(args []string, commandEnv *CommandEnv, writer i continue } fmt.Fprintf(writer, "checking master %s to %s ... ", string(sourceMaster), string(targetMaster)) - err := pb.WithMasterClient(false, sourceMaster, commandEnv.option.GrpcDialOption, false, func(client master_pb.SeaweedClient) error { + err := pb.WithMasterClient(context.Background(), false, sourceMaster, commandEnv.option.GrpcDialOption, false, func(client master_pb.SeaweedClient) error { pong, err := client.Ping(context.Background(), &master_pb.PingRequest{ Target: string(targetMaster), TargetType: cluster.MasterType, diff --git a/weed/shell/command_s3_iam_client.go b/weed/shell/command_s3_iam_client.go index 2b4cd3ce7..3bf405fd1 100644 --- a/weed/shell/command_s3_iam_client.go +++ b/weed/shell/command_s3_iam_client.go @@ -24,7 +24,7 @@ const iamRequestTimeout = 30 * time.Second // applied — callers can derive child contexts but should not need their own // timeout boilerplate. func (ce *CommandEnv) withIamClient(fn func(ctx context.Context, client iam_pb.SeaweedIdentityAccessManagementClient) error) error { - return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error { + return pb.WithGrpcClient(context.Background(), false, 0, func(conn *grpc.ClientConn) error { ctx, cancel := context.WithTimeout(iamAdminAuthContext(context.Background()), iamRequestTimeout) defer cancel() return fn(ctx, iam_pb.NewSeaweedIdentityAccessManagementClient(conn)) diff --git a/weed/shell/s3tables_helpers.go b/weed/shell/s3tables_helpers.go index 0c58c6ac8..94dab9d92 100644 --- a/weed/shell/s3tables_helpers.go +++ b/weed/shell/s3tables_helpers.go @@ -17,7 +17,7 @@ const s3TablesDefaultRegion = "" const timeFormat = "2006-01-02T15:04:05Z07:00" func withFilerClient(commandEnv *CommandEnv, fn func(client filer_pb.SeaweedFilerClient) error) error { - return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error { + return pb.WithGrpcClient(context.Background(), false, 0, func(conn *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(conn) return fn(client) }, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption) diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 8d9cb695d..935e7b90c 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -461,7 +461,7 @@ func (s *Store) cachedLookupEcShardLocations(ecVolume *erasure_coding.EcVolume) glog.V(3).Infof("lookup and cache ec volume %d locations", ecVolume.VolumeId) - err = operation.WithMasterServerClient(false, s.MasterAddress, s.grpcDialOption, func(masterClient master_pb.SeaweedClient) error { + err = operation.WithMasterServerClient(context.Background(), false, s.MasterAddress, s.grpcDialOption, func(masterClient master_pb.SeaweedClient) error { req := &master_pb.LookupEcVolumeRequest{ VolumeId: uint32(ecVolume.VolumeId), } diff --git a/weed/wdclient/masterclient.go b/weed/wdclient/masterclient.go index 4dd3a86a4..d6ca48b38 100644 --- a/weed/wdclient/masterclient.go +++ b/weed/wdclient/masterclient.go @@ -75,7 +75,7 @@ func (p *masterVolumeProvider) LookupVolumeIds(ctx context.Context, volumeIds [] return status.Errorf(codes.Unavailable, "no master available") } - return pb.WithMasterClient(false, master, p.masterClient.grpcDialOption, false, func(client master_pb.SeaweedClient) error { + return pb.WithMasterClient(timeoutCtx, false, master, p.masterClient.grpcDialOption, false, func(client master_pb.SeaweedClient) error { resp, err := client.LookupVolume(timeoutCtx, &master_pb.LookupVolumeRequest{ VolumeOrFileIds: volumeIds, }) @@ -224,7 +224,7 @@ func (mc *MasterClient) tryConnectToMaster(ctx context.Context, master pb.Server glog.V(1).Infof("%s.%s masterClient Connecting to master %v", mc.FilerGroup, mc.clientType, master) stats.MasterClientConnectCounter.WithLabelValues("total").Inc() connectStartTime := time.Now() - gprcErr := pb.WithMasterClient(true, master, mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error { + gprcErr := pb.WithMasterClient(context.Background(), true, master, mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error { ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -394,7 +394,7 @@ func (mc *MasterClient) WithClient(streamingMode bool, fn func(client master_pb. func (mc *MasterClient) WithClientCustomGetMaster(getMasterF func() pb.ServerAddress, streamingMode bool, fn func(client master_pb.SeaweedClient) error) error { return util.Retry("master grpc", func() error { - return pb.WithMasterClient(streamingMode, getMasterF(), mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error { + return pb.WithMasterClient(context.Background(), streamingMode, getMasterF(), mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error { return fn(client) }) }) @@ -511,7 +511,7 @@ func (mc *MasterClient) FindLeaderFromOtherPeers(myMasterAddress pb.ServerAddres if master == myMasterAddress { continue } - if grpcErr := pb.WithMasterClient(false, master, mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error { + if grpcErr := pb.WithMasterClient(context.Background(), false, master, mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error { ctx, cancel := context.WithTimeout(context.Background(), 120*time.Millisecond) defer cancel() resp, err := client.GetMasterConfiguration(ctx, &master_pb.GetMasterConfigurationRequest{})