diff --git a/weed/operation/grpc_client.go b/weed/operation/grpc_client.go index e9063e4b3..c80a1dc6a 100644 --- a/weed/operation/grpc_client.go +++ b/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...)...) } diff --git a/weed/operation/tail_volume.go b/weed/operation/tail_volume.go index 5538701ed..ba1a9ad37 100644 --- a/weed/operation/tail_volume.go +++ b/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...) } diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index fd8e0deed..b00ca67ef 100644 --- a/weed/server/volume_grpc_copy.go +++ b/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 diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 98a41e755..6cf27013d 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/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) } diff --git a/weed/server/volume_grpc_tail.go b/weed/server/volume_grpc_tail.go index 120675b7f..327e5c49e 100644 --- a/weed/server/volume_grpc_tail.go +++ b/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)) } diff --git a/weed/shell/command_volume_merge.go b/weed/shell/command_volume_merge.go index 296cdbe8c..d107c5cf9 100644 --- a/weed/shell/command_volume_merge.go +++ b/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) }()