operation: fold extra dial options into WithVolumeServerClient (#11392)

WithVolumeServerClientOptions duplicated the existing helper only to
append extra dial options. Trail the extras as a variadic parameter on
WithVolumeServerClient instead, keeping the (dialOption, fn) argument
order so all existing callers keep working. TailVolumeFromSource gets
the same treatment.

weed/operation/grpc_client.go

@@ -10,19 +10,15 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 )
 
-func WithVolumeServerClient(streamingMode bool, volumeServer pb.ServerAddress, grpcDialOption grpc.DialOption, fn func(volume_server_pb.VolumeServerClient) error) error {
-	return WithVolumeServerClientOptions(streamingMode, volumeServer, fn, grpcDialOption)
-}
-
-// WithVolumeServerClientOptions is WithVolumeServerClient with extra dial
-// options appended after the TLS option, so a caller dialing an untrusted
-// source address can pin the validated endpoint at connect time.
-func WithVolumeServerClientOptions(streamingMode bool, volumeServer pb.ServerAddress, fn func(volume_server_pb.VolumeServerClient) error, grpcDialOptions ...grpc.DialOption) error {
+// WithVolumeServerClient dials with the TLS option plus any extra dial options
+// appended after it, so a caller dialing an untrusted source address can pin
+// the validated endpoint at connect time.
+func WithVolumeServerClient(streamingMode bool, volumeServer pb.ServerAddress, grpcDialOption grpc.DialOption, fn func(volume_server_pb.VolumeServerClient) error, extraDialOptions ...grpc.DialOption) 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, grpcDialOptions...)
+	}, volumeServer.ToGrpcAddress(), false, append([]grpc.DialOption{grpcDialOption}, extraDialOptions...)...)
 
 }
 

weed/operation/tail_volume.go

@@ -25,11 +25,11 @@ func TailVolume(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vid needle
 
 	volumeServer := lookup.Locations[0].ServerAddress()
 
-	return TailVolumeFromSource(volumeServer, vid, sinceNs, timeoutSeconds, fn, grpcDialOption)
+	return TailVolumeFromSource(volumeServer, vid, sinceNs, timeoutSeconds, grpcDialOption, fn)
 }
 
-func TailVolumeFromSource(volumeServer pb.ServerAddress, vid needle.VolumeId, sinceNs uint64, idleTimeoutSeconds int, fn func(n *needle.Needle) error, grpcDialOptions ...grpc.DialOption) error {
-	return WithVolumeServerClientOptions(true, volumeServer, func(client volume_server_pb.VolumeServerClient) error {
+func TailVolumeFromSource(volumeServer pb.ServerAddress, vid needle.VolumeId, sinceNs uint64, idleTimeoutSeconds int, grpcDialOption grpc.DialOption, fn func(n *needle.Needle) error, extraDialOptions ...grpc.DialOption) error {
+	return WithVolumeServerClient(true, volumeServer, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
 
@@ -90,5 +90,5 @@ func TailVolumeFromSource(volumeServer pb.ServerAddress, vid needle.VolumeId, si
 
 		}
 		return nil
-	}, grpcDialOptions...)
+	}, extraDialOptions...)
 }

weed/server/volume_grpc_copy.go

@@ -59,7 +59,7 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre
 	var sourceVolumeStatusAfterCopy *volume_server_pb.VolumeStatusResponse
 	var dataBaseFileName, indexBaseFileName, idxFileName, datFileName string
 	var hasRemoteDatFile bool
-	err := operation.WithVolumeServerClientOptions(true, pb.ServerAddress(req.SourceDataNode), func(client volume_server_pb.VolumeServerClient) error {
+	err := operation.WithVolumeServerClient(true, pb.ServerAddress(req.SourceDataNode), vs.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
 		var err error
 		sourceVolumeStatus, err = client.VolumeStatus(stream.Context(), &volume_server_pb.VolumeStatusRequest{
 			VolumeId: req.VolumeId,
@@ -214,7 +214,7 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre
 		}
 
 		return nil
-	}, vs.grpcDialOption, vs.guardedGrpcDialOption(req.SourceDataNode))
+	}, vs.guardedGrpcDialOption(req.SourceDataNode))
 
 	if err != nil {
 		return err

weed/server/volume_grpc_erasure_coding.go

@@ -392,7 +392,7 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv
 	}
 	throttler := util.NewWriteThrottler(ioBytePerSecond)
 
-	err := operation.WithVolumeServerClientOptions(true, pb.ServerAddress(req.SourceDataNode), func(client volume_server_pb.VolumeServerClient) error {
+	err := operation.WithVolumeServerClient(true, pb.ServerAddress(req.SourceDataNode), vs.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
 
 		// copy ec data slices
 		for _, shardId := range req.ShardIds {
@@ -460,7 +460,7 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv
 			}
 		}
 		return nil
-	}, vs.grpcDialOption, vs.guardedGrpcDialOption(req.SourceDataNode))
+	}, vs.guardedGrpcDialOption(req.SourceDataNode))
 	if err != nil {
 		return nil, fmt.Errorf("VolumeEcShardsCopy volume %d: %v", req.VolumeId, err)
 	}

weed/server/volume_grpc_tail.go

@@ -100,10 +100,10 @@ func (vs *VolumeServer) VolumeTailReceiver(ctx context.Context, req *volume_serv
 
 	defer glog.V(1).Infof("receive tailing volume %d finished", v.Id)
 
-	return resp, operation.TailVolumeFromSource(pb.ServerAddress(req.SourceVolumeServer), v.Id, req.SinceNs, int(req.IdleTimeoutSeconds), func(n *needle.Needle) error {
+	return resp, operation.TailVolumeFromSource(pb.ServerAddress(req.SourceVolumeServer), v.Id, req.SinceNs, int(req.IdleTimeoutSeconds), vs.grpcDialOption, func(n *needle.Needle) error {
 		_, err := vs.store.WriteVolumeNeedle(v.Id, n, false, false)
 		return err
-	}, vs.grpcDialOption, vs.guardedGrpcDialOption(req.SourceVolumeServer))
+	}, vs.guardedGrpcDialOption(req.SourceVolumeServer))
 
 }
 

weed/shell/command_volume_merge.go

@@ -232,14 +232,14 @@ func startTailNeedleStream(grpcDialOption grpc.DialOption, volumeId needle.Volum
 	ch := make(chan *needle.Needle, 32)
 	stream := &tailNeedleStream{ch: ch}
 	go func() {
-		err := operation.TailVolumeFromSource(server, volumeId, 0, mergeIdleTimeoutSeconds, func(n *needle.Needle) error {
+		err := operation.TailVolumeFromSource(server, volumeId, 0, mergeIdleTimeoutSeconds, grpcDialOption, func(n *needle.Needle) error {
 			select {
 			case ch <- n:
 			case <-done:
 				return fmt.Errorf("merge cancelled")
 			}
 			return nil
-		}, grpcDialOption)
+		})
 		close(ch)
 		stream.setErr(err)
 	}()