diff --git a/test/erasure_coding/chaos_lifecycle_test.go b/test/erasure_coding/chaos_lifecycle_test.go index 68678b914..7fb1002b1 100644 --- a/test/erasure_coding/chaos_lifecycle_test.go +++ b/test/erasure_coding/chaos_lifecycle_test.go @@ -933,7 +933,7 @@ func copyFileContents(src, dst string) error { // chaosDataNodes lists the data nodes from a fresh master topology snapshot. func chaosDataNodes(commandEnv *shell.CommandEnv) []*master_pb.DataNodeInfo { var resp *master_pb.VolumeListResponse - err := commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err := commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { var e error resp, e = client.VolumeList(context.Background(), &master_pb.VolumeListRequest{}) return e @@ -955,7 +955,7 @@ func chaosDataNodes(commandEnv *shell.CommandEnv) []*master_pb.DataNodeInfo { func masterEcGenerations(commandEnv *shell.CommandEnv, volumeId uint32) map[int64]bool { generations := map[int64]bool{} var resp *master_pb.VolumeListResponse - err := commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err := commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { var e error resp, e = client.VolumeList(context.Background(), &master_pb.VolumeListRequest{}) return e diff --git a/test/erasure_coding/multidisk_shardloss_test.go b/test/erasure_coding/multidisk_shardloss_test.go index c58c6ec3d..dc86b5bad 100644 --- a/test/erasure_coding/multidisk_shardloss_test.go +++ b/test/erasure_coding/multidisk_shardloss_test.go @@ -200,7 +200,7 @@ func disksWithShards(testDir string, volumeId uint32) int { func nodeVolumeDiskCounts(t *testing.T, commandEnv *shell.CommandEnv) map[string]int { t.Helper() var resp *master_pb.VolumeListResponse - err := commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err := commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { var e error resp, e = client.VolumeList(context.Background(), &master_pb.VolumeListRequest{}) return e diff --git a/test/erasure_coding/multidisk_shell_lifecycle_test.go b/test/erasure_coding/multidisk_shell_lifecycle_test.go index a2f1ea7de..3a9408ddd 100644 --- a/test/erasure_coding/multidisk_shell_lifecycle_test.go +++ b/test/erasure_coding/multidisk_shell_lifecycle_test.go @@ -370,7 +370,7 @@ func removeTwoShardFiles(t *testing.T, testDir string, volumeId uint32) []int { func masterEcShardIds(commandEnv *shell.CommandEnv, volumeId uint32) map[int]bool { ids := map[int]bool{} var resp *master_pb.VolumeListResponse - err := commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err := commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { var e error resp, e = client.VolumeList(context.Background(), &master_pb.VolumeListRequest{}) return e diff --git a/weed/admin/dash/client_management.go b/weed/admin/dash/client_management.go index 46948a628..6de7d1901 100644 --- a/weed/admin/dash/client_management.go +++ b/weed/admin/dash/client_management.go @@ -20,7 +20,7 @@ import ( // WithMasterClient executes a function with a master client connection func (s *AdminServer) WithMasterClient(f func(client master_pb.SeaweedClient) error) error { - return s.masterClient.WithClient(false, f) + return s.masterClient.WithClient(context.Background(), false, f) } // WithFilerClient executes a function with a filer client connection diff --git a/weed/credential/propagating_store.go b/weed/credential/propagating_store.go index f561d2007..8d42feb47 100644 --- a/weed/credential/propagating_store.go +++ b/weed/credential/propagating_store.go @@ -55,7 +55,7 @@ func (s *PropagatingCredentialStore) propagateChange(ctx context.Context, fn fun // List S3 servers var s3Servers []string - err := s.masterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err := s.masterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error { glog.V(4).Infof("IAM: listing S3 servers (FilerGroup: '%s')", s.masterClient.FilerGroup) resp, err := client.ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{ ClientType: cluster.S3Type, diff --git a/weed/filer/filer_delete_entry.go b/weed/filer/filer_delete_entry.go index 207fdc7e6..cbffcced0 100644 --- a/weed/filer/filer_delete_entry.go +++ b/weed/filer/filer_delete_entry.go @@ -63,7 +63,9 @@ func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isR if isDeleteCollection { collectionName := entry.Name() - f.DoDeleteCollection(collectionName) + // the entry is already gone: a caller that hung up must not leave the + // collection behind, so this cleanup outlives the request + f.DoDeleteCollection(context.Background(), collectionName) // drop bucket-labeled series held by this process; the S3 gateway // only cleans its own registry stats.DeleteBucketMetrics(collectionName) @@ -167,10 +169,10 @@ func (f *Filer) doDeleteEntryMetaAndData(ctx context.Context, entry *Entry, shou return nil } -func (f *Filer) DoDeleteCollection(collectionName string) (err error) { +func (f *Filer) DoDeleteCollection(ctx context.Context, collectionName string) (err error) { - return f.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { - _, err := client.CollectionDelete(context.Background(), &master_pb.CollectionDeleteRequest{ + return f.MasterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error { + _, err := client.CollectionDelete(ctx, &master_pb.CollectionDeleteRequest{ Name: collectionName, }) if err != nil { diff --git a/weed/mq/kafka/integration/seaweedmq_handler_utils.go b/weed/mq/kafka/integration/seaweedmq_handler_utils.go index 843b72280..33c3590ff 100644 --- a/weed/mq/kafka/integration/seaweedmq_handler_utils.go +++ b/weed/mq/kafka/integration/seaweedmq_handler_utils.go @@ -54,7 +54,7 @@ func NewSeaweedMQBrokerHandler(masters string, filerGroup string, clientHost str // Discover brokers from masters using master client glog.V(1).Infof("About to call discoverBrokersWithMasterClient...") - brokerAddresses, err := discoverBrokersWithMasterClient(masterClient, filerGroup) + brokerAddresses, err := discoverBrokersWithMasterClient(ctx, masterClient, filerGroup) if err != nil { glog.Errorf("Broker discovery failed: %v", err) return nil, fmt.Errorf("failed to discover brokers: %v", err) @@ -66,7 +66,7 @@ func NewSeaweedMQBrokerHandler(masters string, filerGroup string, clientHost str } // Discover filers from masters using master client - filerAddresses, err := discoverFilersWithMasterClient(masterClient, filerGroup) + filerAddresses, err := discoverFilersWithMasterClient(ctx, masterClient, filerGroup) if err != nil { return nil, fmt.Errorf("failed to discover filers: %v", err) } @@ -107,12 +107,12 @@ func NewSeaweedMQBrokerHandler(masters string, filerGroup string, clientHost str } // discoverBrokersWithMasterClient queries masters for available brokers using reusable master client -func discoverBrokersWithMasterClient(masterClient *wdclient.MasterClient, filerGroup string) ([]string, error) { +func discoverBrokersWithMasterClient(ctx context.Context, masterClient *wdclient.MasterClient, filerGroup string) ([]string, error) { var brokers []string - err := masterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err := masterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error { glog.V(1).Infof("Inside MasterClient.WithClient callback - client obtained successfully") - resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{ + resp, err := client.ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{ ClientType: cluster.BrokerType, FilerGroup: filerGroup, Limit: 1000, @@ -144,11 +144,11 @@ func discoverBrokersWithMasterClient(masterClient *wdclient.MasterClient, filerG } // discoverFilersWithMasterClient queries masters for available filers using reusable master client -func discoverFilersWithMasterClient(masterClient *wdclient.MasterClient, filerGroup string) ([]pb.ServerAddress, error) { +func discoverFilersWithMasterClient(ctx context.Context, masterClient *wdclient.MasterClient, filerGroup string) ([]pb.ServerAddress, error) { var filers []pb.ServerAddress - err := masterClient.WithClient(false, func(client master_pb.SeaweedClient) error { - resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{ + err := masterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error { + resp, err := client.ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{ ClientType: cluster.FilerType, FilerGroup: filerGroup, Limit: 1000, diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index 9bc9f9a7a..0454396ba 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -890,8 +890,8 @@ func (fs *FilerServer) CollectionList(ctx context.Context, req *filer_pb.Collect glog.V(4).InfofCtx(ctx, "CollectionList %v", req) resp = &filer_pb.CollectionListResponse{} - err = fs.filer.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { - masterResp, err := client.CollectionList(context.Background(), &master_pb.CollectionListRequest{ + err = fs.filer.MasterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error { + masterResp, err := client.CollectionList(ctx, &master_pb.CollectionListRequest{ IncludeNormalVolumes: req.IncludeNormalVolumes, IncludeEcVolumes: req.IncludeEcVolumes, }) @@ -911,7 +911,7 @@ func (fs *FilerServer) DeleteCollection(ctx context.Context, req *filer_pb.Delet glog.V(4).InfofCtx(ctx, "DeleteCollection %v", req) - err = fs.filer.DoDeleteCollection(req.GetCollection()) + err = fs.filer.DoDeleteCollection(ctx, req.GetCollection()) return &filer_pb.DeleteCollectionResponse{}, err } diff --git a/weed/server/filer_grpc_server_admin.go b/weed/server/filer_grpc_server_admin.go index 0cef9e0c4..c013cff2e 100644 --- a/weed/server/filer_grpc_server_admin.go +++ b/weed/server/filer_grpc_server_admin.go @@ -21,8 +21,8 @@ func (fs *FilerServer) Statistics(ctx context.Context, req *filer_pb.StatisticsR var output *master_pb.StatisticsResponse - err = fs.filer.MasterClient.WithClient(false, func(masterClient master_pb.SeaweedClient) error { - grpcResponse, grpcErr := masterClient.Statistics(context.Background(), &master_pb.StatisticsRequest{ + err = fs.filer.MasterClient.WithClient(ctx, false, func(masterClient master_pb.SeaweedClient) error { + grpcResponse, grpcErr := masterClient.Statistics(ctx, &master_pb.StatisticsRequest{ Replication: fs.statisticsReplication(req.Replication), Collection: req.Collection, Ttl: req.Ttl, diff --git a/weed/shell/command_cluster_check.go b/weed/shell/command_cluster_check.go index ca94ab7be..18e4232a3 100644 --- a/weed/shell/command_cluster_check.go +++ b/weed/shell/command_cluster_check.go @@ -62,7 +62,7 @@ func (c *commandClusterCheck) Do(args []string, commandEnv *CommandEnv, writer i // collect filers var filers []pb.ServerAddress - err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{ ClientType: cluster.FilerType, FilerGroup: *commandEnv.option.FilerGroup, diff --git a/weed/shell/command_cluster_ps.go b/weed/shell/command_cluster_ps.go index a2fa8dc19..0c1e04fa8 100644 --- a/weed/shell/command_cluster_ps.go +++ b/weed/shell/command_cluster_ps.go @@ -49,7 +49,7 @@ func (c *commandClusterPs) Do(args []string, commandEnv *CommandEnv, writer io.W listClusterNodes := func(clientType string) ([]*master_pb.ListClusterNodesResponse_ClusterNode, error) { var nodes []*master_pb.ListClusterNodesResponse_ClusterNode - err := commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err := commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{ ClientType: clientType, FilerGroup: *commandEnv.option.FilerGroup, diff --git a/weed/shell/command_cluster_raft_add.go b/weed/shell/command_cluster_raft_add.go index a957d7c44..f54083180 100644 --- a/weed/shell/command_cluster_raft_add.go +++ b/weed/shell/command_cluster_raft_add.go @@ -46,7 +46,7 @@ func (c *commandRaftServerAdd) Do(args []string, commandEnv *CommandEnv, writer return fmt.Errorf("empty server id or address") } - err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { _, err := client.RaftAddServer(context.Background(), &master_pb.RaftAddServerRequest{ Id: *serverId, Address: *serverAddress, diff --git a/weed/shell/command_cluster_raft_leader_transfer.go b/weed/shell/command_cluster_raft_leader_transfer.go index 32dc83f51..0047d8efc 100644 --- a/weed/shell/command_cluster_raft_leader_transfer.go +++ b/weed/shell/command_cluster_raft_leader_transfer.go @@ -67,7 +67,7 @@ func (c *commandRaftLeaderTransfer) Do(args []string, commandEnv *CommandEnv, wr fmt.Fprintf(writer, "Checking current raft cluster status...\n") var currentLeader string - err := commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err := commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() @@ -107,7 +107,7 @@ func (c *commandRaftLeaderTransfer) Do(args []string, commandEnv *CommandEnv, wr } fmt.Fprintf(writer, "\nTransferring leadership from %s to %s...\n", currentLeader, targetDesc) - err = commandEnv.MasterClient.WithClient(true, func(client master_pb.SeaweedClient) error { + err = commandEnv.MasterClient.WithClient(context.Background(), true, func(client master_pb.SeaweedClient) error { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() diff --git a/weed/shell/command_cluster_raft_ps.go b/weed/shell/command_cluster_raft_ps.go index 714e17649..d97272b42 100644 --- a/weed/shell/command_cluster_raft_ps.go +++ b/weed/shell/command_cluster_raft_ps.go @@ -38,7 +38,7 @@ func (c *commandRaftClusterPs) Do(args []string, commandEnv *CommandEnv, writer return nil } - err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { resp, err := client.RaftListClusterServers(context.Background(), &master_pb.RaftListClusterServersRequest{}) if err != nil { return fmt.Errorf("raft list cluster: %w", err) diff --git a/weed/shell/command_cluster_raft_remove.go b/weed/shell/command_cluster_raft_remove.go index 6f5324731..bc18fc05a 100644 --- a/weed/shell/command_cluster_raft_remove.go +++ b/weed/shell/command_cluster_raft_remove.go @@ -44,7 +44,7 @@ func (c *commandRaftServerRemove) Do(args []string, commandEnv *CommandEnv, writ return fmt.Errorf("empty server id") } - err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { _, err := client.RaftRemoveServer(context.Background(), &master_pb.RaftRemoveServerRequest{ Id: *serverId, Force: true, diff --git a/weed/shell/command_collection_delete.go b/weed/shell/command_collection_delete.go index 92473e8e3..889da7f99 100644 --- a/weed/shell/command_collection_delete.go +++ b/weed/shell/command_collection_delete.go @@ -64,7 +64,7 @@ func (c *commandCollectionDelete) Do(args []string, commandEnv *CommandEnv, writ return nil } - err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { _, err = client.CollectionDelete(context.Background(), &master_pb.CollectionDeleteRequest{ Name: *collectionName, }) diff --git a/weed/shell/command_collection_list.go b/weed/shell/command_collection_list.go index 0287b1e96..4114f73d6 100644 --- a/weed/shell/command_collection_list.go +++ b/weed/shell/command_collection_list.go @@ -79,7 +79,7 @@ func (c *commandCollectionList) Do(args []string, commandEnv *CommandEnv, writer func ListCollectionNames(commandEnv *CommandEnv, includeNormalVolumes, includeEcVolumes bool) (collections []string, err error) { var resp *master_pb.CollectionListResponse - err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { resp, err = client.CollectionList(context.Background(), &master_pb.CollectionListRequest{ IncludeNormalVolumes: includeNormalVolumes, IncludeEcVolumes: includeEcVolumes, diff --git a/weed/shell/command_ec_common.go b/weed/shell/command_ec_common.go index 1fe691522..7922f876b 100644 --- a/weed/shell/command_ec_common.go +++ b/weed/shell/command_ec_common.go @@ -55,7 +55,7 @@ func _getDefaultReplicaPlacement(commandEnv *CommandEnv) (*super_block.ReplicaPl var resp *master_pb.GetMasterConfigurationResponse var err error - err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { resp, err = client.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{}) return err }) @@ -94,7 +94,7 @@ func collectTopologyInfo(commandEnv *CommandEnv, delayBeforeCollecting time.Dura } var resp *master_pb.VolumeListResponse - err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { resp, err = pb.CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{}) return err }) diff --git a/weed/shell/command_ec_decode.go b/weed/shell/command_ec_decode.go index 9a10fbf95..33e640d18 100644 --- a/weed/shell/command_ec_decode.go +++ b/weed/shell/command_ec_decode.go @@ -124,7 +124,7 @@ func (c *commandEcDecode) Do(args []string, commandEnv *CommandEnv, writer io.Wr func lookupVolumeIds(commandEnv *CommandEnv, volumeIds []string) (volumeIdLocations []*master_pb.LookupVolumeResponse_VolumeIdLocation, err error) { var resp *master_pb.LookupVolumeResponse - err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { resp, err = client.LookupVolume(context.Background(), &master_pb.LookupVolumeRequest{VolumeOrFileIds: volumeIds}) return err }) diff --git a/weed/shell/command_fs_merge_volumes.go b/weed/shell/command_fs_merge_volumes.go index 6a888cff5..2156c6f3a 100644 --- a/weed/shell/command_fs_merge_volumes.go +++ b/weed/shell/command_fs_merge_volumes.go @@ -318,7 +318,7 @@ func (c *commandFsMergeVolumes) volumesAreCompatible(src needle.VolumeId, dest n func (c *commandFsMergeVolumes) reloadVolumesInfo(masterClient *wdclient.MasterClient) error { c.volumes = make(map[needle.VolumeId]*master_pb.VolumeInformationMessage) - return masterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + return masterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { volumes, err := pb.CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{}) if err != nil { return err diff --git a/weed/shell/command_s3_bucket_delete.go b/weed/shell/command_s3_bucket_delete.go index ddd3201e9..19ea4f329 100644 --- a/weed/shell/command_s3_bucket_delete.go +++ b/weed/shell/command_s3_bucket_delete.go @@ -66,7 +66,7 @@ func (c *commandS3BucketDelete) Do(args []string, commandEnv *CommandEnv, writer } // delete the collection directly first - err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err = commandEnv.MasterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error { _, err = client.CollectionDelete(ctx, &master_pb.CollectionDeleteRequest{ Name: getCollectionName(commandEnv, *bucketName), }) diff --git a/weed/shell/command_volume_grow.go b/weed/shell/command_volume_grow.go index 1a9a043d7..240506deb 100644 --- a/weed/shell/command_volume_grow.go +++ b/weed/shell/command_volume_grow.go @@ -115,7 +115,7 @@ func (c *commandGrow) Do(args []string, commandEnv *CommandEnv, writer io.Writer if !collectionFound { return fmt.Errorf("collection not found") } - if err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + if err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { if _, err := client.VolumeGrow(context.Background(), volumeGrowRequest); err != nil { return err } diff --git a/weed/shell/command_volume_vacuum.go b/weed/shell/command_volume_vacuum.go index 34f7870d9..4a761f2b5 100644 --- a/weed/shell/command_volume_vacuum.go +++ b/weed/shell/command_volume_vacuum.go @@ -99,7 +99,7 @@ func (c *commandVacuum) Do(args []string, commandEnv *CommandEnv, writer io.Writ } for _, volumeId := range volumeIdInts { - err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { _, err = client.VacuumVolume(context.Background(), &master_pb.VacuumVolumeRequest{ GarbageThreshold: float32(*garbageThreshold), VolumeId: volumeId, diff --git a/weed/shell/command_volume_vacuum_disable.go b/weed/shell/command_volume_vacuum_disable.go index 15897ddb0..436e963ff 100644 --- a/weed/shell/command_volume_vacuum_disable.go +++ b/weed/shell/command_volume_vacuum_disable.go @@ -36,7 +36,7 @@ func (c *commandDisableVacuum) Do(args []string, commandEnv *CommandEnv, writer return } - err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { _, err = client.DisableVacuum(context.Background(), &master_pb.DisableVacuumRequest{}) return err }) diff --git a/weed/shell/command_volume_vacuum_enable.go b/weed/shell/command_volume_vacuum_enable.go index b8f779b14..0d5104e9d 100644 --- a/weed/shell/command_volume_vacuum_enable.go +++ b/weed/shell/command_volume_vacuum_enable.go @@ -36,7 +36,7 @@ func (c *commandEnableVacuum) Do(args []string, commandEnv *CommandEnv, writer i return } - err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { _, err = client.EnableVacuum(context.Background(), &master_pb.EnableVacuumRequest{}) return err }) diff --git a/weed/shell/commands.go b/weed/shell/commands.go index 896a2d81b..315dbcb58 100644 --- a/weed/shell/commands.go +++ b/weed/shell/commands.go @@ -112,11 +112,11 @@ func (ce *CommandEnv) isLocked() bool { // shellLockHolder asks the master who currently holds the cluster-wide shell // lock. Best effort: masters without GetAdminLockStatus report no holder, and -// each attempt is bounded so an unresponsive master cannot hang the shell. +// the lookup is bounded so an unresponsive master cannot hang the shell. func (ce *CommandEnv) shellLockHolder() (clientName string, message string, held bool) { - ce.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { - attemptCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) - defer cancel() + attemptCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + ce.MasterClient.WithClient(attemptCtx, false, func(client master_pb.SeaweedClient) error { resp, err := client.GetAdminLockStatus(attemptCtx, &master_pb.GetAdminLockStatusRequest{ LockName: cluster.AdminShellLockName, }) diff --git a/weed/shell/shell_liner.go b/weed/shell/shell_liner.go index aaeaf6ae2..1ae4ce5c5 100644 --- a/weed/shell/shell_liner.go +++ b/weed/shell/shell_liner.go @@ -61,7 +61,7 @@ func RunShell(options ShellOptions) { if commandEnv.option.FilerAddress == "" { var filers []pb.ServerAddress - commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + commandEnv.MasterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error { resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{ ClientType: cluster.FilerType, FilerGroup: *options.FilerGroup, diff --git a/weed/wdclient/exclusive_locks/exclusive_locker.go b/weed/wdclient/exclusive_locks/exclusive_locker.go index e45775165..b1cd94f8a 100644 --- a/weed/wdclient/exclusive_locks/exclusive_locker.go +++ b/weed/wdclient/exclusive_locks/exclusive_locker.go @@ -61,7 +61,7 @@ func (l *ExclusiveLocker) RequestLock(clientName string) { // retry to get the lease for { - if err := l.masterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + if err := l.masterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { attemptCtx, cancel := context.WithTimeout(context.Background(), rpcTimeout) defer cancel() resp, err := client.LeaseAdminToken(attemptCtx, &master_pb.LeaseAdminTokenRequest{ @@ -118,7 +118,7 @@ func (l *ExclusiveLocker) renewLease(ctx context.Context) error { if !l.isLocked.Load() { return nil } - return l.masterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + return l.masterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error { attemptCtx, cancel := context.WithTimeout(ctx, rpcTimeout) defer cancel() resp, err := client.LeaseAdminToken(attemptCtx, &master_pb.LeaseAdminTokenRequest{ @@ -152,7 +152,7 @@ func (l *ExclusiveLocker) ReleaseLock() { // single unbounded attempt: a release cut short by a deadline leaves the // lock held until it expires, turning a slow unlock into a ghost lock - l.masterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + l.masterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error { client.ReleaseAdminToken(ctx, &master_pb.ReleaseAdminTokenRequest{ PreviousToken: prevToken, PreviousLockTime: prevLockTsNs, diff --git a/weed/wdclient/masterclient.go b/weed/wdclient/masterclient.go index b06dffad0..4fe8a1e2a 100644 --- a/weed/wdclient/masterclient.go +++ b/weed/wdclient/masterclient.go @@ -409,16 +409,24 @@ func (mc *MasterClient) updateVidMap(resp *master_pb.KeepConnectedResponse) { len(resp.VolumeLocation.NewEcVids), len(resp.VolumeLocation.DeletedEcVids)) } -func (mc *MasterClient) WithClient(streamingMode bool, fn func(client master_pb.SeaweedClient) error) error { +func (mc *MasterClient) WithClient(ctx context.Context, streamingMode bool, fn func(client master_pb.SeaweedClient) error) error { getMasterF := func() pb.ServerAddress { - return mc.GetMaster(context.Background()) + return mc.GetMaster(ctx) } - return mc.WithClientCustomGetMaster(getMasterF, streamingMode, fn) + return mc.WithClientCustomGetMaster(ctx, getMasterF, streamingMode, fn) } -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(context.Background(), streamingMode, getMasterF(), mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error { +// WithClientCustomGetMaster bounds the wait for a master leader by ctx, so a +// caller with a deadline is not parked for the length of an election. The dial +// still gets context.Background(): fn brings its own RPC context, so nothing +// here can attribute a cancellation to the shared connection. +func (mc *MasterClient) WithClientCustomGetMaster(ctx context.Context, getMasterF func() pb.ServerAddress, streamingMode bool, fn func(client master_pb.SeaweedClient) error) error { + return util.RetryWithBackoff(ctx, "master grpc", util.RetryWaitTime, util.IsTransientError, func() error { + master := getMasterF() + if err := ctx.Err(); err != nil { + return err + } + return pb.WithMasterClient(context.Background(), streamingMode, master, mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error { return fn(client) }) }) diff --git a/weed/wdclient/masterclient_test.go b/weed/wdclient/masterclient_test.go index e19ce9e80..c6450ce64 100644 --- a/weed/wdclient/masterclient_test.go +++ b/weed/wdclient/masterclient_test.go @@ -2,11 +2,14 @@ package wdclient import ( "context" + "errors" "testing" "time" "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" ) // TestWaitUntilConnectedWithoutKeepConnected verifies that WaitUntilConnected @@ -101,3 +104,83 @@ func TestMasterClientFilerGroupLogging(t *testing.T) { t.Errorf("Expected clientType %s, got %s", clientType, mc.clientType) } } + +// TestWithClientStopsWaitingOnCanceledContext verifies that WithClient hands the +// caller's context to the wait for a master leader, so a caller that has already +// given up is not parked until an election finishes. +func TestWithClientStopsWaitingOnCanceledContext(t *testing.T) { + mc := NewMasterClient(grpc.EmptyDialOption{}, "test-group", "test-client", "", "", "", pb.ServerDiscovery{}) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + called := false + start := time.Now() + err := mc.WithClient(ctx, false, func(client master_pb.SeaweedClient) error { + called = true + return nil + }) + elapsed := time.Since(start) + + if !errors.Is(err, context.Canceled) { + t.Errorf("expected context.Canceled, got %v", err) + } + if called { + t.Error("callback ran without a master leader") + } + if elapsed > time.Second { + t.Errorf("WithClient blocked for %v with an already canceled context", elapsed) + } +} + +// TestWithClientStopsWaitingOnDeadline verifies the same bound applies to a +// deadline the caller set rather than an outright cancellation. +func TestWithClientStopsWaitingOnDeadline(t *testing.T) { + mc := NewMasterClient(grpc.EmptyDialOption{}, "test-group", "test-client", "", "", "", pb.ServerDiscovery{}) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + start := time.Now() + err := mc.WithClient(ctx, false, func(client master_pb.SeaweedClient) error { + return nil + }) + elapsed := time.Since(start) + + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("expected context.DeadlineExceeded, got %v", err) + } + if elapsed > time.Second { + t.Errorf("WithClient blocked for %v past a 100ms deadline", elapsed) + } +} + +// TestWithClientStopsBackoffOnCancel verifies that a cancellation arriving while +// the retry is backing off cuts the backoff short rather than sleeping it out. +func TestWithClientStopsBackoffOnCancel(t *testing.T) { + mc := NewMasterClient(grpc.WithTransportCredentials(insecure.NewCredentials()), "test-group", "test-client", "", "", "", pb.ServerDiscovery{}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + getMasterF := func() pb.ServerAddress { return "localhost:19333" } + + attempts := 0 + start := time.Now() + err := mc.WithClientCustomGetMaster(ctx, getMasterF, false, func(client master_pb.SeaweedClient) error { + attempts++ + cancel() + return errors.New("connection reset by peer") + }) + elapsed := time.Since(start) + + if attempts != 1 { + t.Errorf("expected 1 attempt, got %d", attempts) + } + if !errors.Is(err, context.Canceled) { + t.Errorf("expected context.Canceled, got %v", err) + } + if elapsed > time.Second { + t.Errorf("WithClientCustomGetMaster slept %v after the context was canceled", elapsed) + } +}