Make long gRPC timeout configurable

app/cmd/add_replica.go

@@ -41,6 +41,12 @@ func AddReplicaCmd() cli.Command {
 				Value:    5,
 				Usage:    "HTTP client timeout for replica file sync server",
 			},
+			cli.IntFlag{
+				Name:     "grpc-timeout-seconds",
+				Required: false,
+				Value:    0,
+				Usage:    "Specify the gRPC timeout for replica rebuilding. If specify a value <= 0, we will use 24h timeout",
+			},
 			cli.StringFlag{
 				Name:     "replica-instance-name",
 				Required: false,
@@ -92,11 +98,12 @@ func addReplica(c *cli.Context) error {
 
 	fastSync := c.Bool("fast-sync")
 	fileSyncHTTPClientTimeout := c.Int("file-sync-http-client-timeout")
+	grpcTimeoutSeconds := c.Int64("grpc-timeout-seconds")
 
 	if c.Bool("restore") {
 		return task.AddRestoreReplica(volumeSize, volumeCurrentSize, replica, replicaInstanceName)
 	}
-	return task.AddReplica(volumeSize, volumeCurrentSize, replica, replicaInstanceName, fileSyncHTTPClientTimeout, fastSync)
+	return task.AddReplica(volumeSize, volumeCurrentSize, replica, replicaInstanceName, fileSyncHTTPClientTimeout, fastSync, grpcTimeoutSeconds)
 }
 
 func StartWithReplicasCmd() cli.Command {

app/cmd/snapshot.go

@@ -171,6 +171,12 @@ func SnapshotCloneCmd() cli.Command {
 				Value:    5,
 				Usage:    "HTTP client timeout for replica file sync server",
 			},
+			cli.IntFlag{
+				Name:     "grpc-timeout-seconds",
+				Required: false,
+				Value:    0,
+				Usage:    "Specify the gRPC timeout for snapshot clone. If specify a value <= 0, we will use 24h timeout",
+			},
 		},
 		Action: func(c *cli.Context) {
 			if err := cloneSnapshot(c); err != nil {
@@ -451,6 +457,7 @@ func cloneSnapshot(c *cli.Context) error {
 	}
 	exportBackingImageIfExist := c.Bool("export-backing-image-if-exist")
 	fileSyncHTTPClientTimeout := c.Int("file-sync-http-client-timeout")
+	grpcTimeoutSeconds := c.Int64("grpc-timeout-seconds")
 
 	controllerClient, err := getControllerClient(c)
 	if err != nil {
@@ -469,7 +476,7 @@ func cloneSnapshot(c *cli.Context) error {
 	defer fromControllerClient.Close()
 
 	if err := sync.CloneSnapshot(controllerClient, fromControllerClient, volumeName, fromVolumeName,
-		snapshotName, exportBackingImageIfExist, fileSyncHTTPClientTimeout); err != nil {
+		snapshotName, exportBackingImageIfExist, fileSyncHTTPClientTimeout, grpcTimeoutSeconds); err != nil {
 		return err
 	}
 	return nil

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/longhorn/go-common-libs v0.0.0-20240605094403-68743dacbcb1
 	github.com/longhorn/go-iscsi-helper v0.0.0-20240427164656-e9439c0018ce
 	github.com/longhorn/sparse-tools v0.0.0-20240427164751-a7b9f1b2c8a8
-	github.com/longhorn/types v0.0.0-20240605091135-ef450e1c04cd
+	github.com/longhorn/types v0.0.0-20240612122407-553c71ad6514
 	github.com/moby/moby v24.0.9+incompatible
 	github.com/pkg/errors v0.9.1
 	github.com/rancher/go-fibmap v0.0.0-20160418233256-5fc9f8c1ed47
@@ -21,8 +21,8 @@ require (
 	github.com/urfave/cli v1.22.15
 	golang.org/x/net v0.24.0
 	golang.org/x/sys v0.20.0
-	google.golang.org/grpc v1.63.2
-	google.golang.org/protobuf v1.34.0
+	google.golang.org/grpc v1.64.0
+	google.golang.org/protobuf v1.34.2
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c
 	gopkg.in/cheggaaa/pb.v2 v2.0.7
 	k8s.io/mount-utils v0.30.0
@@ -65,7 +65,7 @@ require (
 	github.com/slok/goresilience v0.2.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	golang.org/x/text v0.14.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20240227224415-6ceb2ff114de // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 // indirect
 	gopkg.in/VividCortex/ewma.v1 v1.1.1 // indirect
 	gopkg.in/fatih/color.v1 v1.7.0 // indirect
 	gopkg.in/mattn/go-colorable.v0 v0.1.0 // indirect

go.sum

@@ -74,8 +74,8 @@ github.com/longhorn/go-iscsi-helper v0.0.0-20240427164656-e9439c0018ce h1:PxKniE
 github.com/longhorn/go-iscsi-helper v0.0.0-20240427164656-e9439c0018ce/go.mod h1:d9t3gtE+UPjescbCFluXd4xBc8OQT/JrC2cdkk2IXWQ=
 github.com/longhorn/sparse-tools v0.0.0-20240427164751-a7b9f1b2c8a8 h1:lwtmZEomiv8uchwo9JIyoo+lK8J3cLCm7/qzpn6wmzo=
 github.com/longhorn/sparse-tools v0.0.0-20240427164751-a7b9f1b2c8a8/go.mod h1:pvlUkVwRGojXhcTkkzksOe4i7GVk59P2PbJjHIB2Yj0=
-github.com/longhorn/types v0.0.0-20240605091135-ef450e1c04cd h1:gzvHnEc4vdHmOtxwgjC/7YmChbzDsfYiY0wpI3RgB1A=
-github.com/longhorn/types v0.0.0-20240605091135-ef450e1c04cd/go.mod h1:1oEh1cnDDqNSuFh/dH/lvJ3Ssq83SOweTAAPLRY4PMI=
+github.com/longhorn/types v0.0.0-20240612122407-553c71ad6514 h1:IUAe0Lw2Wz5GrUznw6Sp65qA8J0tWi9qVIIuFZZF5O0=
+github.com/longhorn/types v0.0.0-20240612122407-553c71ad6514/go.mod h1:fonrC6SwGpvt+YVlfJ3xMmg0MlOH94T1Qx2+ZEy8n3U=
 github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I=
 github.com/mattn/go-colorable v0.1.4 h1:snbPLB8fVfU9iwbbo30TPtbLRzwWu6aJS6Xh4eaaviA=
 github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
@@ -201,12 +201,12 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20240227224415-6ceb2ff114de h1:cZGRis4/ot9uVm639a+rHCUaG0JJHEsdyzSQTMX+suY=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20240227224415-6ceb2ff114de/go.mod h1:H4O17MA/PE9BsGx3w+a+W2VOLLD1Qf7oJneAoU6WktY=
-google.golang.org/grpc v1.63.2 h1:MUeiw1B2maTVZthpU5xvASfTh3LDbxHd6IJ6QQVU+xM=
-google.golang.org/grpc v1.63.2/go.mod h1:WAX/8DgncnokcFUldAxq7GeB5DXHDbMF+lLvDomNkRA=
-google.golang.org/protobuf v1.34.0 h1:Qo/qEd2RZPCf2nKuorzksSknv0d3ERwp1vFG38gSmH4=
-google.golang.org/protobuf v1.34.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 h1:NnYq6UN9ReLM9/Y01KWNOWyI5xQ9kbIms5GGJVwS/Yc=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237/go.mod h1:WtryC6hu0hhx87FDGxWCDptyssuo68sk10vYjF+T9fY=
+google.golang.org/grpc v1.64.0 h1:KH3VH9y/MgNQg1dE7b3XfVK0GsPSIzJwdF617gUSbvY=
+google.golang.org/grpc v1.64.0/go.mod h1:oxjF8E3FBnjp+/gVFYdWacaLDx9na1aqy9oovLpxQYg=
+google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
+google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
 gopkg.in/VividCortex/ewma.v1 v1.1.1 h1:tWHEKkKq802K/JT9RiqGCBU5fW3raAPnJGTE9ostZvg=
 gopkg.in/VividCortex/ewma.v1 v1.1.1/go.mod h1:TekXuFipeiHWiAlO1+wSS23vTcyFau5u3rxXUSXj710=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

pkg/controller/rebuild.go

@@ -118,7 +118,7 @@ func syncFile(from, to, fromAddress, toAddress, volumeName, toInstanceName strin
 	strHostPort := net.JoinHostPort(host, strconv.Itoa(int(port)))
 
 	logrus.Infof("Synchronizing %s to %s:%s", from, to, strHostPort)
-	err = fromClient.SendFile(from, host, port, fileSyncHTTPClientTimeout, fastSync)
+	err = fromClient.SendFile(from, host, port, fileSyncHTTPClientTimeout, fastSync, 0)
 	if err != nil {
 		logrus.WithError(err).Errorf("failed to synchronize %s to %s:%s", from, to, strHostPort)
 	} else {

pkg/replica/client/client.go

@@ -442,12 +442,16 @@ func (c *ReplicaClient) RenameFile(oldFileName, newFileName string) error {
 	return nil
 }
 
-func (c *ReplicaClient) SendFile(from, host string, port int32, fileSyncHTTPClientTimeout int, fastSync bool) error {
+func (c *ReplicaClient) SendFile(from, host string, port int32, fileSyncHTTPClientTimeout int, fastSync bool, grpcTimeoutSeconds int64) error {
 	syncAgentServiceClient, err := c.getSyncServiceClient()
 	if err != nil {
 		return err
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), GRPCServiceLongTimeout)
+	grpcTimeout := GRPCServiceLongTimeout
+	if grpcTimeoutSeconds > 0 {
+		grpcTimeout = time.Second * time.Duration(grpcTimeoutSeconds)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), grpcTimeout)
 	defer cancel()
 
 	if _, err := syncAgentServiceClient.FileSend(ctx, &enginerpc.FileSendRequest{
@@ -502,12 +506,16 @@ func (c *ReplicaClient) LaunchReceiver(toFilePath string) (string, int32, error)
 	return c.host, reply.Port, nil
 }
 
-func (c *ReplicaClient) SyncFiles(fromAddress string, list []types.SyncFileInfo, fileSyncHTTPClientTimeout int, fastSync bool) error {
+func (c *ReplicaClient) SyncFiles(fromAddress string, list []types.SyncFileInfo, fileSyncHTTPClientTimeout int, fastSync bool, grpcTimeoutSeconds int64) error {
 	syncAgentServiceClient, err := c.getSyncServiceClient()
 	if err != nil {
 		return err
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), GRPCServiceLongTimeout)
+	grpcTimeout := GRPCServiceLongTimeout
+	if grpcTimeoutSeconds > 0 {
+		grpcTimeout = time.Second * time.Duration(grpcTimeoutSeconds)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), grpcTimeout)
 	defer cancel()
 
 	if _, err := syncAgentServiceClient.FilesSync(ctx, &enginerpc.FilesSyncRequest{
@@ -516,6 +524,7 @@ func (c *ReplicaClient) SyncFiles(fromAddress string, list []types.SyncFileInfo,
 		SyncFileInfoList:          syncFileInfoListToSyncAgentGRPCFormat(list),
 		FastSync:                  fastSync,
 		FileSyncHttpClientTimeout: int32(fileSyncHTTPClientTimeout),
+		GrpcTimeoutSeconds:        grpcTimeoutSeconds,
 	}); err != nil {
 		return errors.Wrapf(err, "failed to sync files %+v from %v", list, fromAddress)
 	}
@@ -686,12 +695,16 @@ func (c *ReplicaClient) ReplicaRebuildStatus() (*enginerpc.ReplicaRebuildStatusR
 	return status, nil
 }
 
-func (c *ReplicaClient) CloneSnapshot(fromAddress, fromVolumeName, snapshotFileName string, exportBackingImageIfExist bool, fileSyncHTTPClientTimeout int) error {
+func (c *ReplicaClient) CloneSnapshot(fromAddress, fromVolumeName, snapshotFileName string, exportBackingImageIfExist bool, fileSyncHTTPClientTimeout int, grpcTimeoutSeconds int64) error {
 	syncAgentServiceClient, err := c.getSyncServiceClient()
 	if err != nil {
 		return err
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), GRPCServiceLongTimeout)
+	grpcTimeout := GRPCServiceLongTimeout
+	if grpcTimeoutSeconds > 0 {
+		grpcTimeout = time.Second * time.Duration(grpcTimeoutSeconds)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), grpcTimeout)
 	defer cancel()
 
 	if _, err := syncAgentServiceClient.SnapshotClone(ctx, &enginerpc.SnapshotCloneRequest{

pkg/sync/rpc/server.go

@@ -467,7 +467,7 @@ func (s *SyncAgentServer) FilesSync(ctx context.Context, req *enginerpc.FilesSyn
 		if err != nil {
 			return nil, errors.Wrapf(err, "failed to launch receiver for file %v", info.ToFileName)
 		}
-		if err := fromClient.SendFile(info.FromFileName, req.ToHost, int32(port), int(req.FileSyncHttpClientTimeout), req.FastSync); err != nil {
+		if err := fromClient.SendFile(info.FromFileName, req.ToHost, int32(port), int(req.FileSyncHttpClientTimeout), req.FastSync, req.GrpcTimeoutSeconds); err != nil {
 			return nil, errors.Wrapf(err, "replica %v failed to send file %v to %v:%v", req.FromAddress, info.ToFileName, req.ToHost, port)
 		}
 	}

pkg/sync/sync.go

@@ -412,7 +412,7 @@ func (t *Task) VerifyRebuildReplica(address, instanceName string) error {
 	return nil
 }
 
-func (t *Task) AddReplica(volumeSize, volumeCurrentSize int64, address, instanceName string, fileSyncHTTPClientTimeout int, fastSync bool) error {
+func (t *Task) AddReplica(volumeSize, volumeCurrentSize int64, address, instanceName string, fileSyncHTTPClientTimeout int, fastSync bool, grpcTimeoutSeconds int64) error {
 	volume, err := t.client.VolumeGet()
 	if err != nil {
 		return err
@@ -459,7 +459,7 @@ func (t *Task) AddReplica(volumeSize, volumeCurrentSize int64, address, instance
 		return fmt.Errorf("sync file list shouldn't contain volume head")
 	}
 
-	if err = toClient.SyncFiles(fromAddress, resp, fileSyncHTTPClientTimeout, fastSync); err != nil {
+	if err = toClient.SyncFiles(fromAddress, resp, fileSyncHTTPClientTimeout, fastSync, grpcTimeoutSeconds); err != nil {
 		return err
 	}
 
@@ -808,7 +808,7 @@ func (t *Task) RebuildStatus() (map[string]*ReplicaRebuildStatus, error) {
 }
 
 func CloneSnapshot(engineControllerClient, fromControllerClient *client.ControllerClient, volumeName, fromVolumeName,
-	snapshotFileName string, exportBackingImageIfExist bool, fileSyncHTTPClientTimeout int) error {
+	snapshotFileName string, exportBackingImageIfExist bool, fileSyncHTTPClientTimeout int, grpcTimeoutSeconds int64) error {
 	replicas, err := fromControllerClient.ReplicaList()
 	if err != nil {
 		return err
@@ -851,7 +851,7 @@ func CloneSnapshot(engineControllerClient, fromControllerClient *client.Controll
 			}
 			defer repClient.Close()
 			if err := repClient.CloneSnapshot(sourceReplica.Address, fromVolumeName, snapshotFileName,
-				exportBackingImageIfExist, fileSyncHTTPClientTimeout); err != nil {
+				exportBackingImageIfExist, fileSyncHTTPClientTimeout, grpcTimeoutSeconds); err != nil {
 				syncErrorMap.Store(r.Address, err)
 			}
 		}(r)