Implement volume resizing

pkg/client/client.go

@@ -542,6 +542,22 @@ func (c *SPDKClient) EngineReplicaDelete(engineName, replicaName, replicaAddress
 	return errors.Wrapf(err, "failed to delete replica %s with address %s to engine %s", replicaName, replicaAddress, engineName)
 }
 
+func (c *SPDKClient) EngineVolumeResize(engineName string, size uint64) error {
+	if engineName == "" {
+		return fmt.Errorf("failed to resize volume for SPDK engine: missing required parameter engine name")
+	}
+
+	client := c.getSPDKServiceClient()
+	ctx, cancel := context.WithTimeout(context.Background(), GRPCServiceTimeout)
+	defer cancel()
+
+	_, err := client.EngineVolumeResize(ctx, &spdkrpc.EngineVolumeResizeRequest{
+		EngineName: engineName,
+		Size:       size,
+	})
+	return errors.Wrapf(err, "failed to resize volume for engine %s", engineName)
+}
+
 func (c *SPDKClient) EngineBackupCreate(req *BackupCreateRequest) (*spdkrpc.BackupCreateResponse, error) {
 	client := c.getSPDKServiceClient()
 	ctx, cancel := context.WithTimeout(context.Background(), GRPCServiceTimeout)

pkg/spdk/engine.go

@@ -169,6 +169,63 @@ func (e *Engine) Create(spdkClient *spdkclient.Client, replicaAddressMap, localR
 	return e.getWithoutLock(), nil
 }
 
+func (e *Engine) resizeVolume(spdkClient *spdkclient.Client, size uint64) (err error) {
+	e.Lock()
+	defer e.Unlock()
+
+	if e.State != types.InstanceStateRunning {
+		return fmt.Errorf("invalid state %s for engine %s resize bdev", e.State, e.Name)
+	}
+
+	roundedSize := util.RoundUp(size, helpertypes.MiB)
+	if roundedSize != size {
+		e.log.Infof("Rounded up spec size from %v to %v since the spec size should be multiple of MiB", size, roundedSize)
+	}
+	e.log.WithField("specSize", size)
+
+	if e.SpecSize == roundedSize {
+		return nil
+	}
+
+	if roundedSize < e.ActualSize {
+		return fmt.Errorf("cannot shrink the engine %s bdev size from %d to %d", e.Name, e.ActualSize, size)
+	}
+
+	// Now, we need to resize each replica
+	for replicaName, replicaAddress := range e.ReplicaAddressMap {
+		if e.ReplicaModeMap[replicaName] != types.ModeRW {
+			continue
+		}
+		if err := e.resizeReplica(spdkClient, replicaName, replicaAddress, roundedSize); err != nil {
+			return err
+		}
+	}
+
+	e.SpecSize = roundedSize
+	e.log.Infof("Resized engine %s bdev size to %d", e.Name, e.SpecSize)
+	return nil
+}
+
+func (e *Engine) resizeReplica(spdkClient *spdkclient.Client, replicaName, replicaAddress string, size uint64) (err error) {
+	replicaIP, _, err := net.SplitHostPort(replicaAddress)
+	if err != nil {
+		return err
+	}
+	if replicaIP == e.IP {
+		return nil
+	}
+
+	bdevName := e.ReplicaBdevNameMap[replicaName]
+	if bdevName == "" {
+		return nil
+	}
+	if _, err := spdkClient.BdevLvolResize(bdevName, size); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func (e *Engine) getBdevNameForReplica(spdkClient *spdkclient.Client, localReplicaLvsNameMap map[string]string, replicaName, replicaAddress, podIP string) (bdevName string, err error) {
 	replicaIP, _, err := net.SplitHostPort(replicaAddress)
 	if err != nil {

pkg/spdk/server.go

@@ -1144,6 +1144,21 @@ func (s *Server) EngineRestoreStatus(ctx context.Context, req *spdkrpc.RestoreSt
 	return resp, nil
 }
 
+func (s *Server) EngineVolumeResize(ctx context.Context, req *spdkrpc.EngineVolumeResizeRequest) (ret *emptypb.Empty, err error) {
+	s.RLock()
+	e := s.engineMap[req.EngineName]
+	s.RUnlock()
+
+	if e == nil {
+		return nil, grpcstatus.Errorf(grpccodes.NotFound, "cannot find engine %v for volume resize", req.EngineName)
+	}
+
+	if err := e.resizeVolume(s.spdkClient, req.Size); err != nil {
+		return nil, err
+	}
+	return &emptypb.Empty{}, nil
+}
+
 func (s *Server) ReplicaRestoreStatus(ctx context.Context, req *spdkrpc.ReplicaRestoreStatusRequest) (ret *spdkrpc.ReplicaRestoreStatusResponse, err error) {
 	s.RLock()
 	defer s.RUnlock()