diff --git a/weed/shell/command_ec_encode.go b/weed/shell/command_ec_encode.go index 596af9eaf..68fae93b6 100644 --- a/weed/shell/command_ec_encode.go +++ b/weed/shell/command_ec_encode.go @@ -403,7 +403,7 @@ func doEcEncode(commandEnv *CommandEnv, writer io.Writer, volumeIdToCollection m for _, vid := range volumeIds { for _, l := range locations[vid] { ewg.Add(func() error { - if err := markVolumeReplicaWritable(commandEnv.option.GrpcDialOption, vid, l, false, false); err != nil { + if err := markVolumeReplicaWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, l, false, false); err != nil { return fmt.Errorf("mark volume %d as readonly on %s: %v", vid, l.Url, err) } return nil @@ -902,7 +902,7 @@ func doDeleteVolumesWithLocations(commandEnv *CommandEnv, volumeIds []needle.Vol for _, l := range locations { ewg.Add(func() error { - if err := deleteVolume(commandEnv.option.GrpcDialOption, vid, l.ServerAddress(), false, false); err != nil { + if err := deleteVolume(context.Background(), commandEnv.option.GrpcDialOption, vid, l.ServerAddress(), false, false); err != nil { return fmt.Errorf("deleteVolume %s volume %d: %v", l.Url, vid, err) } fmt.Printf("deleted volume %d from %s\n", vid, l.Url) diff --git a/weed/shell/command_volume_balance.go b/weed/shell/command_volume_balance.go index ff8f521a0..ed7222f42 100644 --- a/weed/shell/command_volume_balance.go +++ b/weed/shell/command_volume_balance.go @@ -2,15 +2,17 @@ package shell import ( "cmp" + "context" "flag" "fmt" - "github.com/seaweedfs/seaweedfs/weed/util" "io" "os" "regexp" "strings" "time" + "github.com/seaweedfs/seaweedfs/weed/util" + "slices" "github.com/seaweedfs/seaweedfs/weed/pb" @@ -655,7 +657,7 @@ func moveVolume(commandEnv *CommandEnv, v *master_pb.VolumeInformationMessage, f } fmt.Fprintf(os.Stdout, " moving %s volume %s%d %s => %s\n", v.DiskType, collectionPrefix, v.Id, fullNode.info.Id, emptyNode.info.Id) if applyChange { - return LiveMoveVolume(commandEnv.option.GrpcDialOption, os.Stderr, needle.VolumeId(v.Id), pb.NewServerAddressFromDataNode(fullNode.info), pb.NewServerAddressFromDataNode(emptyNode.info), 5*time.Second, v.DiskType, 0, v.ReadOnly) + return LiveMoveVolume(context.Background(), commandEnv.option.GrpcDialOption, os.Stderr, needle.VolumeId(v.Id), pb.NewServerAddressFromDataNode(fullNode.info), pb.NewServerAddressFromDataNode(emptyNode.info), 5*time.Second, v.DiskType, 0, v.ReadOnly) } return nil } diff --git a/weed/shell/command_volume_copy.go b/weed/shell/command_volume_copy.go index 9cbabdb23..6fe486751 100644 --- a/weed/shell/command_volume_copy.go +++ b/weed/shell/command_volume_copy.go @@ -1,6 +1,7 @@ package shell import ( + "context" "flag" "fmt" "io" @@ -60,6 +61,6 @@ func (c *commandVolumeCopy) Do(args []string, commandEnv *CommandEnv, writer io. return fmt.Errorf("source and target volume servers are the same!") } - _, err = copyVolume(commandEnv.option.GrpcDialOption, writer, volumeId, sourceVolumeServer, targetVolumeServer, "", 0, true) + _, err = copyVolume(context.Background(), commandEnv.option.GrpcDialOption, writer, volumeId, sourceVolumeServer, targetVolumeServer, "", 0, true) return } diff --git a/weed/shell/command_volume_delete.go b/weed/shell/command_volume_delete.go index 975a7a909..f69a2997a 100644 --- a/weed/shell/command_volume_delete.go +++ b/weed/shell/command_volume_delete.go @@ -1,6 +1,7 @@ package shell import ( + "context" "flag" "io" @@ -50,6 +51,6 @@ func (c *commandVolumeDelete) Do(args []string, commandEnv *CommandEnv, writer i volumeId := needle.VolumeId(*volumeIdInt) - return deleteVolume(commandEnv.option.GrpcDialOption, volumeId, sourceVolumeServer, false, false) + return deleteVolume(context.Background(), commandEnv.option.GrpcDialOption, volumeId, sourceVolumeServer, false, false) } diff --git a/weed/shell/command_volume_delete_empty.go b/weed/shell/command_volume_delete_empty.go index 0c12a033c..2ff79a248 100644 --- a/weed/shell/command_volume_delete_empty.go +++ b/weed/shell/command_volume_delete_empty.go @@ -1,6 +1,7 @@ package shell import ( + "context" "flag" "io" "log" @@ -72,7 +73,7 @@ func (c *commandVolumeDeleteEmpty) Do(args []string, commandEnv *CommandEnv, wri if isEmptyVolumeDeleteCandidate(v, quietSeconds, nowUnixSeconds, *collectionPattern) { if *applyBalancing { log.Printf("deleting empty volume %d from %s", v.Id, dn.Id) - if deleteErr := deleteVolume(commandEnv.option.GrpcDialOption, needle.VolumeId(v.Id), + if deleteErr := deleteVolume(context.Background(), commandEnv.option.GrpcDialOption, needle.VolumeId(v.Id), pb.NewServerAddressFromDataNode(dn), true, false); deleteErr != nil { err = deleteErr } diff --git a/weed/shell/command_volume_fix_replication.go b/weed/shell/command_volume_fix_replication.go index ee89dfc17..21e05d583 100644 --- a/weed/shell/command_volume_fix_replication.go +++ b/weed/shell/command_volume_fix_replication.go @@ -1,6 +1,7 @@ package shell import ( + "context" "flag" "fmt" "io" @@ -396,7 +397,7 @@ func (c *commandVolumeFixReplication) deleteOneVolume(commandEnv *CommandEnv, wr // Surplus replica being trimmed; keep the remote object since other // replicas of the same .vif still reference it. - if err := deleteVolume(commandEnv.option.GrpcDialOption, needle.VolumeId(replica.info.Id), + if err := deleteVolume(context.Background(), commandEnv.option.GrpcDialOption, needle.VolumeId(replica.info.Id), pb.NewServerAddressFromDataNode(replica.location.dataNode), false, true); err != nil { fmt.Fprintf(writer, "deleting volume %d from %s : %v", replica.info.Id, replica.location.dataNode.Id, err) } else { @@ -538,7 +539,7 @@ func (c *commandVolumeFixReplication) fixOneUnderReplicatedVolume(commandEnv *Co return true, nil } - err := replicateVolumeToServer(commandEnv.option.GrpcDialOption, writer, needle.VolumeId(replica.info.Id), + err := replicateVolumeToServer(context.Background(), commandEnv.option.GrpcDialOption, writer, needle.VolumeId(replica.info.Id), pb.NewServerAddressFromDataNode(replica.location.dataNode), pb.NewServerAddressFromDataNode(dst.dataNode), replica.info.DiskType) diff --git a/weed/shell/command_volume_fsck.go b/weed/shell/command_volume_fsck.go index 120885145..eb75e30c9 100644 --- a/weed/shell/command_volume_fsck.go +++ b/weed/shell/command_volume_fsck.go @@ -496,7 +496,7 @@ func (c *commandVolumeFsck) findExtraChunksInVolumeServers(dataNodeVolumeIdToVIn // purgeOneVolume picks the orphan fids to delete for a single volume and // fires the delete RPC. It's split out of findExtraChunksInVolumeServers so -// the `defer markVolumeWritable(..., false, false)` at the bottom fires +// the `defer markVolumeWritable(context.Background(), ..., false, false)` at the bottom fires // between volumes — putting that defer inside the caller's for-loop would // leave every processed volume writable until the whole fsck run finished. func (c *commandVolumeFsck) purgeOneVolume(volumeId uint32, orphanReplicaFileIds map[string]int, replicaCount int, readOnlyReplicas []pb.ServerAddress) error { @@ -518,12 +518,12 @@ func (c *commandVolumeFsck) purgeOneVolume(volumeId uint32, orphanReplicaFileIds needleVID := needle.VolumeId(volumeId) for _, server := range readOnlyReplicas { - if err := markVolumeWritable(c.env.option.GrpcDialOption, needleVID, server, true, false); err != nil { + if err := markVolumeWritable(context.Background(), c.env.option.GrpcDialOption, needleVID, server, true, false); err != nil { // Replicas flipped writable earlier roll back via the defer. return fmt.Errorf("mark %v writable: %v", server, err) } fmt.Fprintf(c.writer, "temporarily marked %d on server %v writable for forced purge\n", volumeId, server) - defer markVolumeWritable(c.env.option.GrpcDialOption, needleVID, server, false, false) + defer markVolumeWritable(context.Background(), c.env.option.GrpcDialOption, needleVID, server, false, false) } if *c.verbose { diff --git a/weed/shell/command_volume_mark.go b/weed/shell/command_volume_mark.go index 39453fcca..80e0e90fb 100644 --- a/weed/shell/command_volume_mark.go +++ b/weed/shell/command_volume_mark.go @@ -1,6 +1,7 @@ package shell import ( + "context" "errors" "flag" "fmt" @@ -96,7 +97,7 @@ func (c *commandVolumeMark) Do(args []string, commandEnv *CommandEnv, writer io. } var failures []error for _, target := range targets { - if err := markVolumeWritable(commandEnv.option.GrpcDialOption, target.volumeId, target.sourceVolumeServer, markWritable, true); err != nil { + if err := markVolumeWritable(context.Background(), commandEnv.option.GrpcDialOption, target.volumeId, target.sourceVolumeServer, markWritable, true); err != nil { failures = append(failures, fmt.Errorf("mark volume %d on %s: %w", target.volumeId, target.sourceVolumeServer, err)) fmt.Fprintf(writer, "volume %d on %s: %v\n", target.volumeId, target.sourceVolumeServer, err) continue @@ -113,7 +114,7 @@ func (c *commandVolumeMark) Do(args []string, commandEnv *CommandEnv, writer io. volumeId := needle.VolumeId(*volumeIdInt) - return markVolumeWritable(commandEnv.option.GrpcDialOption, volumeId, sourceVolumeServer, markWritable, true) + return markVolumeWritable(context.Background(), commandEnv.option.GrpcDialOption, volumeId, sourceVolumeServer, markWritable, true) } type volumeMarkTarget struct { diff --git a/weed/shell/command_volume_merge.go b/weed/shell/command_volume_merge.go index 06c4407d8..c115af20d 100644 --- a/weed/shell/command_volume_merge.go +++ b/weed/shell/command_volume_merge.go @@ -120,7 +120,7 @@ func (c *commandVolumeMerge) Do(args []string, commandEnv *CommandEnv, writer io if !cleanupTarget { return } - if delErr := deleteVolume(commandEnv.option.GrpcDialOption, volumeId, targetServer, false, false); delErr != nil { + if delErr := deleteVolume(context.Background(), commandEnv.option.GrpcDialOption, volumeId, targetServer, false, false); delErr != nil { glog.Warningf("failed to clean up temporary merge volume %d on %s: %v", volumeId, targetServer, delErr) } }() @@ -187,12 +187,12 @@ func (c *commandVolumeMerge) Do(args []string, commandEnv *CommandEnv, writer io for i, replica := range replicas { sourceServer := pb.NewServerAddressFromDataNode(replica.location.dataNode) - if _, err = copyVolume(commandEnv.option.GrpcDialOption, writer, volumeId, targetServer, sourceServer, "", 0, false); err != nil { + if _, err = copyVolume(context.Background(), commandEnv.option.GrpcDialOption, writer, volumeId, targetServer, sourceServer, "", 0, false); err != nil { return fmt.Errorf("rebuild replica %d/%d on %s from merged volume %d: %w; merged copy kept on %s, re-run to finish", i+1, len(replicas), sourceServer, volumeId, err, targetServer) } } - if err = deleteVolume(commandEnv.option.GrpcDialOption, volumeId, targetServer, false, false); err != nil { + if err = deleteVolume(context.Background(), commandEnv.option.GrpcDialOption, volumeId, targetServer, false, false); err != nil { return err } diff --git a/weed/shell/command_volume_move.go b/weed/shell/command_volume_move.go index 1f3214af4..541aa98b7 100644 --- a/weed/shell/command_volume_move.go +++ b/weed/shell/command_volume_move.go @@ -83,20 +83,20 @@ func (c *commandVolumeMove) Do(args []string, commandEnv *CommandEnv, writer io. return fmt.Errorf("source and target volume servers are the same!") } - return LiveMoveVolume(commandEnv.option.GrpcDialOption, writer, volumeId, sourceVolumeServer, targetVolumeServer, 5*time.Second, *diskTypeStr, *ioBytePerSecond, false) + return LiveMoveVolume(context.Background(), commandEnv.option.GrpcDialOption, writer, volumeId, sourceVolumeServer, targetVolumeServer, 5*time.Second, *diskTypeStr, *ioBytePerSecond, false) } // LiveMoveVolume moves one volume from one source volume server to one target volume server, with idleTimeout to drain the incoming requests. -func LiveMoveVolume(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceVolumeServer, targetVolumeServer pb.ServerAddress, idleTimeout time.Duration, diskType string, ioBytePerSecond int64, skipTailError bool) (err error) { +func LiveMoveVolume(ctx context.Context, grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceVolumeServer, targetVolumeServer pb.ServerAddress, idleTimeout time.Duration, diskType string, ioBytePerSecond int64, skipTailError bool) (err error) { log.Printf("copying volume %d from %s to %s", volumeId, sourceVolumeServer, targetVolumeServer) - lastAppendAtNs, err := copyVolume(grpcDialOption, writer, volumeId, sourceVolumeServer, targetVolumeServer, diskType, ioBytePerSecond, false) + lastAppendAtNs, err := copyVolume(ctx, grpcDialOption, writer, volumeId, sourceVolumeServer, targetVolumeServer, diskType, ioBytePerSecond, false) if err != nil { return fmt.Errorf("copy volume %d from %s to %s: %v", volumeId, sourceVolumeServer, targetVolumeServer, err) } log.Printf("tailing volume %d from %s to %s", volumeId, sourceVolumeServer, targetVolumeServer) - if err = tailVolume(grpcDialOption, volumeId, sourceVolumeServer, targetVolumeServer, lastAppendAtNs, idleTimeout); err != nil { + if err = tailVolume(ctx, grpcDialOption, volumeId, sourceVolumeServer, targetVolumeServer, lastAppendAtNs, idleTimeout); err != nil { if skipTailError { fmt.Fprintf(writer, "tail volume %d from %s to %s: %v\n", volumeId, sourceVolumeServer, targetVolumeServer, err) } else { @@ -105,7 +105,7 @@ func LiveMoveVolume(grpcDialOption grpc.DialOption, writer io.Writer, volumeId n } log.Printf("deleting volume %d from %s", volumeId, sourceVolumeServer) - if err = deleteVolume(grpcDialOption, volumeId, sourceVolumeServer, false, true); err != nil { + if err = deleteVolume(ctx, grpcDialOption, volumeId, sourceVolumeServer, false, true); err != nil { return fmt.Errorf("delete volume %d from %s: %v", volumeId, sourceVolumeServer, err) } @@ -113,7 +113,7 @@ func LiveMoveVolume(grpcDialOption grpc.DialOption, writer io.Writer, volumeId n return nil } -func copyVolume(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceVolumeServer, targetVolumeServer pb.ServerAddress, diskType string, ioBytePerSecond int64, restoreWritable bool) (lastAppendAtNs uint64, err error) { +func copyVolume(ctx context.Context, grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceVolumeServer, targetVolumeServer pb.ServerAddress, diskType string, ioBytePerSecond int64, restoreWritable bool) (lastAppendAtNs uint64, err error) { // check to see if the volume is already read-only and if its not then we need // to mark it as read-only and then before we return we need to undo what we @@ -127,8 +127,13 @@ func copyVolume(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needl return } + // Restoring writability must outlive a cancelled copy, or the source + // is left permanently readonly on abort, as balance_task.go already + // guards against. + restoreCtx, restoreCancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) + defer restoreCancel() clientErr := operation.WithVolumeServerClient(false, sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { - _, writableErr := volumeServerClient.VolumeMarkWritable(context.Background(), &volume_server_pb.VolumeMarkWritableRequest{ + _, writableErr := volumeServerClient.VolumeMarkWritable(restoreCtx, &volume_server_pb.VolumeMarkWritableRequest{ VolumeId: uint32(volumeId), }) return writableErr @@ -139,12 +144,12 @@ func copyVolume(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needl }() err = operation.WithVolumeServerClient(false, sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { - resp, statusErr := volumeServerClient.VolumeStatus(context.Background(), &volume_server_pb.VolumeStatusRequest{ + resp, statusErr := volumeServerClient.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{ VolumeId: uint32(volumeId), }) if statusErr == nil && !resp.IsReadOnly { shouldMarkWritable = true - _, readonlyErr := volumeServerClient.VolumeMarkReadonly(context.Background(), &volume_server_pb.VolumeMarkReadonlyRequest{ + _, readonlyErr := volumeServerClient.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{ VolumeId: uint32(volumeId), Persist: false, }) @@ -157,7 +162,7 @@ func copyVolume(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needl } err = operation.WithVolumeServerClient(true, targetVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { - stream, replicateErr := volumeServerClient.VolumeCopy(context.Background(), &volume_server_pb.VolumeCopyRequest{ + stream, replicateErr := volumeServerClient.VolumeCopy(ctx, &volume_server_pb.VolumeCopyRequest{ VolumeId: uint32(volumeId), SourceDataNode: string(sourceVolumeServer), DiskType: diskType, @@ -188,10 +193,10 @@ func copyVolume(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needl return } -func tailVolume(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceVolumeServer, targetVolumeServer pb.ServerAddress, lastAppendAtNs uint64, idleTimeout time.Duration) (err error) { +func tailVolume(ctx context.Context, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceVolumeServer, targetVolumeServer pb.ServerAddress, lastAppendAtNs uint64, idleTimeout time.Duration) (err error) { return operation.WithVolumeServerClient(true, targetVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { - _, replicateErr := volumeServerClient.VolumeTailReceiver(context.Background(), &volume_server_pb.VolumeTailReceiverRequest{ + _, replicateErr := volumeServerClient.VolumeTailReceiver(ctx, &volume_server_pb.VolumeTailReceiverRequest{ VolumeId: uint32(volumeId), SinceNs: lastAppendAtNs, IdleTimeoutSeconds: uint32(idleTimeout.Seconds()), @@ -205,9 +210,9 @@ func tailVolume(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, source // deleteVolume removes the volume from sourceVolumeServer. When keepRemoteData // is true, the cloud-tier object backing the volume is left intact — used on // the source side of a move where another server is taking over the same .vif. -func deleteVolume(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceVolumeServer pb.ServerAddress, onlyEmpty bool, keepRemoteData bool) (err error) { +func deleteVolume(ctx context.Context, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceVolumeServer pb.ServerAddress, onlyEmpty bool, keepRemoteData bool) (err error) { return operation.WithVolumeServerClient(false, sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { - _, deleteErr := volumeServerClient.VolumeDelete(context.Background(), &volume_server_pb.VolumeDeleteRequest{ + _, deleteErr := volumeServerClient.VolumeDelete(ctx, &volume_server_pb.VolumeDeleteRequest{ VolumeId: uint32(volumeId), OnlyEmpty: onlyEmpty, KeepRemoteData: keepRemoteData, @@ -216,14 +221,14 @@ func deleteVolume(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sour }) } -func markVolumeWritable(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceVolumeServer pb.ServerAddress, writable, persist bool) (err error) { +func markVolumeWritable(ctx context.Context, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceVolumeServer pb.ServerAddress, writable, persist bool) (err error) { return operation.WithVolumeServerClient(false, sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { if writable { - _, err = volumeServerClient.VolumeMarkWritable(context.Background(), &volume_server_pb.VolumeMarkWritableRequest{ + _, err = volumeServerClient.VolumeMarkWritable(ctx, &volume_server_pb.VolumeMarkWritableRequest{ VolumeId: uint32(volumeId), }) } else { - _, err = volumeServerClient.VolumeMarkReadonly(context.Background(), &volume_server_pb.VolumeMarkReadonlyRequest{ + _, err = volumeServerClient.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{ VolumeId: uint32(volumeId), Persist: persist, }) @@ -232,18 +237,18 @@ func markVolumeWritable(grpcDialOption grpc.DialOption, volumeId needle.VolumeId }) } -func markVolumeReplicaWritable(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, location wdclient.Location, writable, persist bool) error { +func markVolumeReplicaWritable(ctx context.Context, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, location wdclient.Location, writable, persist bool) error { if writable { fmt.Printf("markVolumeWritable %d on %s ...\n", volumeId, location.Url) } else { fmt.Printf("markVolumeReadonly %d on %s persist=%v ...\n", volumeId, location.Url, persist) } - return markVolumeWritable(grpcDialOption, volumeId, location.ServerAddress(), writable, persist) + return markVolumeWritable(ctx, grpcDialOption, volumeId, location.ServerAddress(), writable, persist) } -func markVolumeReplicasWritable(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, locations []wdclient.Location, writable, persist bool) error { +func markVolumeReplicasWritable(ctx context.Context, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, locations []wdclient.Location, writable, persist bool) error { for _, location := range locations { - if err := markVolumeReplicaWritable(grpcDialOption, volumeId, location, writable, persist); err != nil { + if err := markVolumeReplicaWritable(ctx, grpcDialOption, volumeId, location, writable, persist); err != nil { return err } } @@ -251,9 +256,9 @@ func markVolumeReplicasWritable(grpcDialOption grpc.DialOption, volumeId needle. } // replicateVolumeToServer copies a volume from sourceAddress to targetAddress via the VolumeCopy gRPC stream. -func replicateVolumeToServer(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceAddress, targetAddress pb.ServerAddress, diskType string) error { +func replicateVolumeToServer(ctx context.Context, grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceAddress, targetAddress pb.ServerAddress, diskType string) error { return operation.WithVolumeServerClient(false, targetAddress, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { - stream, replicateErr := volumeServerClient.VolumeCopy(context.Background(), &volume_server_pb.VolumeCopyRequest{ + stream, replicateErr := volumeServerClient.VolumeCopy(ctx, &volume_server_pb.VolumeCopyRequest{ VolumeId: uint32(volumeId), SourceDataNode: string(sourceAddress), DiskType: diskType, @@ -278,9 +283,9 @@ func replicateVolumeToServer(grpcDialOption grpc.DialOption, writer io.Writer, v } // configureVolumeReplication sets the replication setting on a volume at the given server. -func configureVolumeReplication(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, targetAddress pb.ServerAddress, replicationString string) error { +func configureVolumeReplication(ctx context.Context, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, targetAddress pb.ServerAddress, replicationString string) error { return operation.WithVolumeServerClient(false, targetAddress, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { - resp, configureErr := volumeServerClient.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{ + resp, configureErr := volumeServerClient.VolumeConfigure(ctx, &volume_server_pb.VolumeConfigureRequest{ VolumeId: uint32(volumeId), Replication: replicationString, }) diff --git a/weed/shell/command_volume_tier_move.go b/weed/shell/command_volume_tier_move.go index 0d098756f..fad43853c 100644 --- a/weed/shell/command_volume_tier_move.go +++ b/weed/shell/command_volume_tier_move.go @@ -1,6 +1,7 @@ package shell import ( + "context" "flag" "fmt" "io" @@ -316,7 +317,7 @@ func (c *commandVolumeTierMove) doMoveOneVolume(commandEnv *CommandEnv, writer i } // mark all replicas as read only - if err = markVolumeReplicasWritable(commandEnv.option.GrpcDialOption, vid, locations, false, false); err != nil { + if err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, locations, false, false); err != nil { return fmt.Errorf("mark volume %d as readonly on %s: %v", vid, locations[0].Url, err) } newAddress := pb.NewServerAddressFromDataNode(dst.dataNode) @@ -325,9 +326,9 @@ func (c *commandVolumeTierMove) doMoveOneVolume(commandEnv *CommandEnv, writer i deletedSource := sourceVolumeServer if alreadyPlaced { deletedSource = "" - } else if err = LiveMoveVolume(commandEnv.option.GrpcDialOption, writer, vid, sourceVolumeServer, newAddress, 5*time.Second, toDiskType.ReadableString(), ioBytePerSecond, true); err != nil { + } else if err = LiveMoveVolume(context.Background(), commandEnv.option.GrpcDialOption, writer, vid, sourceVolumeServer, newAddress, 5*time.Second, toDiskType.ReadableString(), ioBytePerSecond, true); err != nil { // mark all replicas as writable - if err = markVolumeReplicasWritable(commandEnv.option.GrpcDialOption, vid, locations, true, false); err != nil { + if err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, locations, true, false); err != nil { glog.Errorf("mark volume %d as writable on %s: %v", vid, locations[0].Url, err) } @@ -336,7 +337,7 @@ func (c *commandVolumeTierMove) doMoveOneVolume(commandEnv *CommandEnv, writer i // If move is successful and replication is not empty, alter moved volume's replication setting if *replicationString != "" { - if err = configureVolumeReplication(commandEnv.option.GrpcDialOption, vid, newAddress, *replicationString); err != nil { + if err = configureVolumeReplication(context.Background(), commandEnv.option.GrpcDialOption, vid, newAddress, *replicationString); err != nil { // LiveMoveVolume already deleted sourceVolumeServer; mark surviving // old replicas writable before aborting so the volume stays accessible. restoreSurvivingReplicasWritable(commandEnv, vid, locations, deletedSource) @@ -360,7 +361,7 @@ func (c *commandVolumeTierMove) doMoveOneVolume(commandEnv *CommandEnv, writer i // stay read-only since we're keeping rather than deleting them. for _, loc := range locations { if preserveServers[loc.Url] { - if markErr := markVolumeWritable(commandEnv.option.GrpcDialOption, vid, loc.ServerAddress(), true, false); markErr != nil { + if markErr := markVolumeWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, loc.ServerAddress(), true, false); markErr != nil { glog.Errorf("mark volume %d as writable on preserved replica %s: %v", vid, loc.Url, markErr) } } @@ -378,7 +379,7 @@ func (c *commandVolumeTierMove) doMoveOneVolume(commandEnv *CommandEnv, writer i } // keepRemoteData=true: remote-tiered replicas share one cloud object, so // deleting a replica must not delete the object the survivors still point at. - if err = deleteVolume(commandEnv.option.GrpcDialOption, vid, loc.ServerAddress(), false, true); err != nil { + if err = deleteVolume(context.Background(), commandEnv.option.GrpcDialOption, vid, loc.ServerAddress(), false, true); err != nil { fmt.Fprintf(writer, "failed to delete volume %d on %s: %v\n", vid, loc.Url, err) } } @@ -392,7 +393,7 @@ func restoreSurvivingReplicasWritable(commandEnv *CommandEnv, vid needle.VolumeI if loc.ServerAddress() == deletedSource { continue } - if markErr := markVolumeWritable(commandEnv.option.GrpcDialOption, vid, loc.ServerAddress(), true, false); markErr != nil { + if markErr := markVolumeWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, loc.ServerAddress(), true, false); markErr != nil { glog.Errorf("mark volume %d as writable on %s: %v", vid, loc.Url, markErr) } } @@ -475,7 +476,7 @@ func (c *commandVolumeTierMove) ensureReplicationFulfilled(commandEnv *CommandEn if replicationString != "" { for _, r := range targetTierReplicas { addr := pb.NewServerAddressFromDataNode(r.location.dataNode) - if configErr := configureVolumeReplication(commandEnv.option.GrpcDialOption, vid, addr, replicationString); configErr != nil { + if configErr := configureVolumeReplication(context.Background(), commandEnv.option.GrpcDialOption, vid, addr, replicationString); configErr != nil { return nil, fmt.Errorf("volume %d: failed to configure replication on existing replica %s: %v", vid, r.location.dataNode.Id, configErr) } } @@ -509,14 +510,14 @@ func (c *commandVolumeTierMove) ensureReplicationFulfilled(commandEnv *CommandEn candidateAddress := pb.NewServerAddressFromDataNode(candidateDst.dataNode) fmt.Fprintf(writer, "volume %d: replicating from %s to %s\n", vid, sourceAddress, candidateDst.dataNode.Id) - if copyErr := replicateVolumeToServer(commandEnv.option.GrpcDialOption, writer, vid, sourceAddress, candidateAddress, toDiskType.ReadableString()); copyErr != nil { + if copyErr := replicateVolumeToServer(context.Background(), commandEnv.option.GrpcDialOption, writer, vid, sourceAddress, candidateAddress, toDiskType.ReadableString()); copyErr != nil { return nil, fmt.Errorf("replicate volume %d to %s: %v", vid, candidateDst.dataNode.Id, copyErr) } // Configure replication on the new replica if an explicit -toReplication was given. // Without it, VolumeCopy already preserves the source's replication from the super block. if replicationString != "" { - if configErr := configureVolumeReplication(commandEnv.option.GrpcDialOption, vid, candidateAddress, replicationString); configErr != nil { + if configErr := configureVolumeReplication(context.Background(), commandEnv.option.GrpcDialOption, vid, candidateAddress, replicationString); configErr != nil { return nil, fmt.Errorf("volume %d: failed to configure replication on %s: %v", vid, candidateDst.dataNode.Id, configErr) } } diff --git a/weed/shell/command_volume_tier_upload.go b/weed/shell/command_volume_tier_upload.go index 68678d264..eda04884c 100644 --- a/weed/shell/command_volume_tier_upload.go +++ b/weed/shell/command_volume_tier_upload.go @@ -131,7 +131,7 @@ func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection str return fmt.Errorf("volume %d not found in collection %s", vid, collection) } - err = markVolumeReplicasWritable(commandEnv.option.GrpcDialOption, vid, existingLocations, false, false) + err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, existingLocations, false, false) if err != nil { return fmt.Errorf("mark volume %d as readonly on %s: %v", vid, existingLocations[0].Url, err) } @@ -153,7 +153,7 @@ func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection str continue } fmt.Fprintf(writer, "replicate remote volume %d metadata from %s to %s\n", vid, existingLocations[0].Url, location.Url) - err = replicateVolumeToServer(commandEnv.option.GrpcDialOption, writer, vid, existingLocations[0].ServerAddress(), location.ServerAddress(), "") + err = replicateVolumeToServer(context.Background(), commandEnv.option.GrpcDialOption, writer, vid, existingLocations[0].ServerAddress(), location.ServerAddress(), "") if err != nil { return fmt.Errorf("replicate volume %d from %s to %s: %v", vid, existingLocations[0].Url, location.Url, err) }