Make long gRPC timeout configurable

PhanLe1010 · PhanLe1010 · commit f9a4d0285303 · 2024-06-11T16:35:16.000-07:00
longhorn-2765

Signed-off-by: Phan Le &lt;phan.le@suse.com&gt;
diff --git a/app/cmd/add_replica.go b/app/cmd/add_replica.go
@@ -92,11 +92,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 {
diff --git a/app/cmd/snapshot.go b/app/cmd/snapshot.go
@@ -469,7 +469,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, 0); err != nil {
 		return err
 	}
 	return nil
diff --git a/go.mod b/go.mod
@@ -28,6 +28,8 @@ require (
 	k8s.io/mount-utils v0.30.0
 )
 
+replace github.com/longhorn/types => ../types
+
 require (
 	github.com/Azure/azure-sdk-for-go/sdk/azcore v0.21.1 // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/internal v0.8.3 // indirect
diff --git a/go.sum b/go.sum
@@ -74,8 +74,6 @@ 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/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=
diff --git a/pkg/controller/rebuild.go b/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 {
diff --git a/pkg/replica/client/client.go b/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{
diff --git a/pkg/sync/rpc/server.go b/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)
 		}
 	}
diff --git a/pkg/sync/sync.go b/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)
diff --git a/vendor/github.com/longhorn/types/pkg/generated/enginerpc/syncagent.pb.go b/vendor/github.com/longhorn/types/pkg/generated/enginerpc/syncagent.pb.go
diff --git a/vendor/modules.txt b/vendor/modules.txt

Original file line number	Diff line number	Diff line change
`@@ -92,11 +92,12 @@ func addReplica(c *cli.Context) error {`
`92`	`92`
`93`	`93`	`fastSync := c.Bool("fast-sync")`
`94`	`94`	`fileSyncHTTPClientTimeout := c.Int("file-sync-http-client-timeout")`
	`95`	`+ grpcTimeoutSeconds := c.Int64("grpc-timeout-seconds")`
`95`	`96`
`96`	`97`	`if c.Bool("restore") {`
`97`	`98`	`return task.AddRestoreReplica(volumeSize, volumeCurrentSize, replica, replicaInstanceName)`
`98`	`99`	`}`
`99`		`- return task.AddReplica(volumeSize, volumeCurrentSize, replica, replicaInstanceName, fileSyncHTTPClientTimeout, fastSync)`
	`100`	`+ return task.AddReplica(volumeSize, volumeCurrentSize, replica, replicaInstanceName, fileSyncHTTPClientTimeout, fastSync, grpcTimeoutSeconds)`
`100`	`101`	`}`
`101`	`102`
`102`	`103`	`func StartWithReplicasCmd() cli.Command {`
Original file line number	Diff line number	Diff line change
`@@ -469,7 +469,7 @@ func cloneSnapshot(c *cli.Context) error {`
`469`	`469`	`defer fromControllerClient.Close()`
`470`	`470`
`471`	`471`	`if err := sync.CloneSnapshot(controllerClient, fromControllerClient, volumeName, fromVolumeName,`
`472`		`- snapshotName, exportBackingImageIfExist, fileSyncHTTPClientTimeout); err != nil {`
	`472`	`+ snapshotName, exportBackingImageIfExist, fileSyncHTTPClientTimeout, 0); err != nil {`
`473`	`473`	`return err`
`474`	`474`	`}`
`475`	`475`	`return nil`
Original file line number	Diff line number	Diff line change
`@@ -28,6 +28,8 @@ require (`
`28`	`28`	`k8s.io/mount-utils v0.30.0`
`29`	`29`	`)`
`30`	`30`
	`31`	`+replace github.com/longhorn/types => ../types`
	`32`	`+`
`31`	`33`	`require (`
`32`	`34`	`github.com/Azure/azure-sdk-for-go/sdk/azcore v0.21.1 // indirect`
`33`	`35`	`github.com/Azure/azure-sdk-for-go/sdk/internal v0.8.3 // indirect`
Original file line number	Diff line number	Diff line change
`@@ -467,7 +467,7 @@ func (s SyncAgentServer) FilesSync(ctx context.Context, req enginerpc.FilesSyn`
`467`	`467`	`if err != nil {`
`468`	`468`	`return nil, errors.Wrapf(err, "failed to launch receiver for file %v", info.ToFileName)`
`469`	`469`	`}`
`470`		`- if err := fromClient.SendFile(info.FromFileName, req.ToHost, int32(port), int(req.FileSyncHttpClientTimeout), req.FastSync); err != nil {`
	`470`	`+ if err := fromClient.SendFile(info.FromFileName, req.ToHost, int32(port), int(req.FileSyncHttpClientTimeout), req.FastSync, req.GrpcTimeoutSeconds); err != nil {`
`471`	`471`	`return nil, errors.Wrapf(err, "replica %v failed to send file %v to %v:%v", req.FromAddress, info.ToFileName, req.ToHost, port)`
`472`	`472`	`}`
`473`	`473`	`}`
Original file line number	Diff line number	Diff line change
`@@ -412,7 +412,7 @@ func (t *Task) VerifyRebuildReplica(address, instanceName string) error {`
`412`	`412`	`return nil`
`413`	`413`	`}`
`414`	`414`
`415`		`-func (t *Task) AddReplica(volumeSize, volumeCurrentSize int64, address, instanceName string, fileSyncHTTPClientTimeout int, fastSync bool) error {`
	`415`	`+func (t *Task) AddReplica(volumeSize, volumeCurrentSize int64, address, instanceName string, fileSyncHTTPClientTimeout int, fastSync bool, grpcTimeoutSeconds int64) error {`
`416`	`416`	`volume, err := t.client.VolumeGet()`
`417`	`417`	`if err != nil {`
`418`	`418`	`return err`
`@@ -459,7 +459,7 @@ func (t *Task) AddReplica(volumeSize, volumeCurrentSize int64, address, instance`
`459`	`459`	`return fmt.Errorf("sync file list shouldn't contain volume head")`
`460`	`460`	`}`
`461`	`461`
`462`		`- if err = toClient.SyncFiles(fromAddress, resp, fileSyncHTTPClientTimeout, fastSync); err != nil {`
	`462`	`+ if err = toClient.SyncFiles(fromAddress, resp, fileSyncHTTPClientTimeout, fastSync, grpcTimeoutSeconds); err != nil {`
`463`	`463`	`return err`
`464`	`464`	`}`
`465`	`465`
`@@ -808,7 +808,7 @@ func (t Task) RebuildStatus() (map[string]ReplicaRebuildStatus, error) {`
`808`	`808`	`}`
`809`	`809`
`810`	`810`	`func CloneSnapshot(engineControllerClient, fromControllerClient *client.ControllerClient, volumeName, fromVolumeName,`
`811`		`- snapshotFileName string, exportBackingImageIfExist bool, fileSyncHTTPClientTimeout int) error {`
	`811`	`+ snapshotFileName string, exportBackingImageIfExist bool, fileSyncHTTPClientTimeout int, grpcTimeoutSeconds int64) error {`
`812`	`812`	`replicas, err := fromControllerClient.ReplicaList()`
`813`	`813`	`if err != nil {`
`814`	`814`	`return err`
`@@ -851,7 +851,7 @@ func CloneSnapshot(engineControllerClient, fromControllerClient *client.Controll`
`851`	`851`	`}`
`852`	`852`	`defer repClient.Close()`
`853`	`853`	`if err := repClient.CloneSnapshot(sourceReplica.Address, fromVolumeName, snapshotFileName,`
`854`		`- exportBackingImageIfExist, fileSyncHTTPClientTimeout); err != nil {`
	`854`	`+ exportBackingImageIfExist, fileSyncHTTPClientTimeout, grpcTimeoutSeconds); err != nil {`
`855`	`855`	`syncErrorMap.Store(r.Address, err)`
`856`	`856`	`}`
`857`	`857`	`}(r)`