storage pools: implement rsync.bwlimit property

When rsync has to be invoked to transfer storage entities (e.g. on local or
remote migration) there is currently no way to place an upper limit on the
amount of socket I/O allowed. This can lead to problems on systems with limited
ram available. Starting with this commit LXD allows setting rsync.bwlimit to
place an upper limit on the amount of socket I/O allowed. This property is
storage pool specific rather than a global daemon config key. In this way,
users can grant important pools more I/O than less important pools.

Closes #2678.

Signed-off-by: Christian Brauner <christian.brauner@ubuntu.com>

doc/api-extensions.md

@@ -254,4 +254,8 @@ and the "source" attribute is added to specify the device on host.
 If "source" is set without a "path", we should assume that "path" will be the same as "source".
 If "path" is set without "source" and "major/minor" isn't set,
 we should assume that "source" will be the same as "path".
-So at least one of them must be set.
+So at least one of them must be set.
+
+## rsync.bwlimit
+When rsync has to be invoked to transfer storage entities setting rsync.bwlimit
+places an upper limit on the amount of socket I/O allowed.

doc/configuration.md

@@ -398,6 +398,7 @@ volume.block.mount\_options     | string    | block based driver (lvm)
 lvm.thinpool\_name              | string    | lvm driver                        | LXDPool           | Thin pool where images and containers are created.
 lvm.use\_thinpool               | bool      | lvm driver                        | true              | Whether the storage pool uses a thinpool for logical volumes.
 lvm.vg\_name                    | string    | lvm driver                        | name of the pool  | Name of the volume group to create.
+rsync.bwlimit                   | string    | -                                 | 0 (no limit)      | Specifies the upper limit to be placed on the socket I/O whenever rsync has to be used to transfer storage entities.
 volume.size                     | string    | appropriate driver                | 0                 | Default volume size
 volume.zfs.remove\_snapshots    | bool      | zfs driver                        | false             | Remove snapshots as needed
 volume.zfs.use\_refquota        | bool      | zfs driver                        | false             | Use refquota instead of quota for space.

doc/storage-backends.md

@@ -44,6 +44,10 @@ When such capabilities aren't available, either because the storage driver doesn
 or because the storage backend of the source and target servers differ,  
 LXD will fallback to using rsync to transfer the individual files instead.
 
+When rsync has to be used LXD allows to specify an upper limit on the amount of
+socket I/O by setting the "rsync.bwlimit" storage pool property to a non-zero
+value.
+
 ## Default storage pool
 There is no concept of a default storage pool in LXD.  
 Instead, the pool to use for the container's root is treated as just another "disk" device in LXD.

lxd/api_1.0.go

@@ -102,6 +102,7 @@ func api10Get(d *Daemon, r *http.Request) Response {
 			"storage_zfs_clone_copy",
 			"unix_device_rename",
 			"storage_lvm_use_thinpool",
+			"storage_rsync_bwlimit",
 		},
 		APIStatus:  "stable",
 		APIVersion: version.APIVersion,

lxd/db_update.go

@@ -576,7 +576,7 @@ func dbUpdateFromV11(currentVersion int, version int, d *Daemon) error {
 			// containers/<container>/snapshots/<snap0>
 			// to
 			// snapshots/<container>/<snap0>
-			output, err := storageRsyncCopy(oldPath, newPath)
+			output, err := rsyncLocalCopy(oldPath, newPath, "")
 			if err != nil {
 				logger.Error(
 					"Failed rsync snapshot",

lxd/migrate.go

@@ -374,11 +374,18 @@ func (s *migrationSourceWs) Do(migrateOp *operation) error {
 		return err
 	}
 
+	bwlimit := ""
 	if *header.Fs != myType {
 		myType = MigrationFSType_RSYNC
 		header.Fs = &myType
 
 		driver, _ = rsyncMigrationSource(s.container, s.containerOnly)
+
+		// Check if this storage pool has a rate limit set for rsync.
+		poolwritable := s.container.Storage().GetStoragePoolWritable()
+		if poolwritable.Config != nil {
+			bwlimit = poolwritable.Config["rsync.bwlimit"]
+		}
 	}
 
 	// All failure paths need to do a few things to correctly handle errors before returning.
@@ -394,7 +401,7 @@ func (s *migrationSourceWs) Do(migrateOp *operation) error {
 		return err
 	}
 
-	err = driver.SendWhileRunning(s.fsConn, migrateOp)
+	err = driver.SendWhileRunning(s.fsConn, migrateOp, bwlimit)
 	if err != nil {
 		return abort(err)
 	}
@@ -515,12 +522,12 @@ func (s *migrationSourceWs) Do(migrateOp *operation) error {
 		 * no reason to do these in parallel. In the future when we're using
 		 * p.haul's protocol, it will make sense to do these in parallel.
 		 */
-		err = RsyncSend(shared.AddSlash(checkpointDir), s.criuConn, nil)
+		err = RsyncSend(shared.AddSlash(checkpointDir), s.criuConn, nil, bwlimit)
 		if err != nil {
 			return abort(err)
 		}
 
-		err = driver.SendAfterCheckpoint(s.fsConn)
+		err = driver.SendAfterCheckpoint(s.fsConn, bwlimit)
 		if err != nil {
 			return abort(err)
 		}

lxd/patches.go

@@ -393,7 +393,7 @@ func upgradeFromStorageTypeBtrfs(name string, d *Daemon, defaultPoolName string,
 					return err
 				}
 
-				output, err := storageRsyncCopy(oldContainerMntPoint, newContainerMntPoint)
+				output, err := rsyncLocalCopy(oldContainerMntPoint, newContainerMntPoint, "")
 				if err != nil {
 					logger.Errorf("Failed to rsync: %s: %s.", output, err)
 					return err
@@ -480,7 +480,7 @@ func upgradeFromStorageTypeBtrfs(name string, d *Daemon, defaultPoolName string,
 						return err
 					}
 
-					output, err := storageRsyncCopy(oldSnapshotMntPoint, newSnapshotMntPoint)
+					output, err := rsyncLocalCopy(oldSnapshotMntPoint, newSnapshotMntPoint, "")
 					if err != nil {
 						logger.Errorf("Failed to rsync: %s: %s.", output, err)
 						return err
@@ -684,7 +684,7 @@ func upgradeFromStorageTypeDir(name string, d *Daemon, defaultPoolName string, d
 			// First try to rename.
 			err := os.Rename(oldContainerMntPoint, newContainerMntPoint)
 			if err != nil {
-				output, err := storageRsyncCopy(oldContainerMntPoint, newContainerMntPoint)
+				output, err := rsyncLocalCopy(oldContainerMntPoint, newContainerMntPoint, "")
 				if err != nil {
 					logger.Errorf("Failed to rsync: %s: %s.", output, err)
 					return err
@@ -731,7 +731,7 @@ func upgradeFromStorageTypeDir(name string, d *Daemon, defaultPoolName string, d
 		if shared.PathExists(oldSnapshotMntPoint) && !shared.PathExists(newSnapshotMntPoint) {
 			err := os.Rename(oldSnapshotMntPoint, newSnapshotMntPoint)
 			if err != nil {
-				output, err := storageRsyncCopy(oldSnapshotMntPoint, newSnapshotMntPoint)
+				output, err := rsyncLocalCopy(oldSnapshotMntPoint, newSnapshotMntPoint, "")
 				if err != nil {
 					logger.Errorf("Failed to rsync: %s: %s.", output, err)
 					return err
@@ -1057,7 +1057,7 @@ func upgradeFromStorageTypeLvm(name string, d *Daemon, defaultPoolName string, d
 				}
 
 				// Use rsync to fill the empty volume.
-				output, err := storageRsyncCopy(oldContainerMntPoint, newContainerMntPoint)
+				output, err := rsyncLocalCopy(oldContainerMntPoint, newContainerMntPoint, "")
 				if err != nil {
 					ctStorage.ContainerDelete(ctStruct)
 					return fmt.Errorf("rsync failed: %s", string(output))
@@ -1211,7 +1211,7 @@ func upgradeFromStorageTypeLvm(name string, d *Daemon, defaultPoolName string, d
 					}
 
 					// Use rsync to fill the empty volume.
-					output, err := storageRsyncCopy(oldSnapshotMntPoint, newSnapshotMntPoint)
+					output, err := rsyncLocalCopy(oldSnapshotMntPoint, newSnapshotMntPoint, "")
 					if err != nil {
 						csStorage.ContainerDelete(csStruct)
 						return fmt.Errorf("rsync failed: %s", string(output))

lxd/rsync.go

@@ -14,7 +14,36 @@ import (
 	"github.com/lxc/lxd/shared/logger"
 )
 
-func rsyncSendSetup(path string) (*exec.Cmd, net.Conn, io.ReadCloser, error) {
+// rsyncCopy copies a directory using rsync (with the --devices option).
+func rsyncLocalCopy(source string, dest string, bwlimit string) (string, error) {
+	err := os.MkdirAll(dest, 0755)
+	if err != nil {
+		return "", err
+	}
+
+	rsyncVerbosity := "-q"
+	if debug {
+		rsyncVerbosity = "-vi"
+	}
+
+	if bwlimit == "" {
+		bwlimit = "0"
+	}
+
+	return shared.RunCommand("rsync",
+		"-a",
+		"-HAX",
+		"--devices",
+		"--delete",
+		"--checksum",
+		"--numeric-ids",
+		"--bwlimit", bwlimit,
+		rsyncVerbosity,
+		shared.AddSlash(source),
+		dest)
+}
+
+func rsyncSendSetup(path string, bwlimit string) (*exec.Cmd, net.Conn, io.ReadCloser, error) {
 	/*
 	 * It's sort of unfortunate, but there's no library call to get a
 	 * temporary name, so we get the file and close it and use its name.
@@ -57,16 +86,21 @@ func rsyncSendSetup(path string) (*exec.Cmd, net.Conn, io.ReadCloser, error) {
 	 * hardcoding that at the other end, so we can just ignore it.
 	 */
 	rsyncCmd := fmt.Sprintf("sh -c \"%s netcat %s\"", execPath, f.Name())
-	cmd := exec.Command(
-		"rsync",
+	if bwlimit == "" {
+		bwlimit = "0"
+	}
+
+	cmd := exec.Command("rsync",
 		"-arvP",
 		"--devices",
 		"--numeric-ids",
 		"--partial",
 		path,
 		"localhost:/tmp/foo",
 		"-e",
-		rsyncCmd)
+		rsyncCmd,
+		"--bwlimit",
+		bwlimit)
 
 	stderr, err := cmd.StderrPipe()
 	if err != nil {
@@ -90,8 +124,8 @@ func rsyncSendSetup(path string) (*exec.Cmd, net.Conn, io.ReadCloser, error) {
 
 // RsyncSend sets up the sending half of an rsync, to recursively send the
 // directory pointed to by path over the websocket.
-func RsyncSend(path string, conn *websocket.Conn, readWrapper func(io.ReadCloser) io.ReadCloser) error {
-	cmd, dataSocket, stderr, err := rsyncSendSetup(path)
+func RsyncSend(path string, conn *websocket.Conn, readWrapper func(io.ReadCloser) io.ReadCloser, bwlimit string) error {
+	cmd, dataSocket, stderr, err := rsyncSendSetup(path, bwlimit)
 	if err != nil {
 		return err
 	}

lxd/storage.go

@@ -112,33 +112,6 @@ func filesystemDetect(path string) (string, error) {
 	}
 }
 
-// storageRsyncCopy copies a directory using rsync (with the --devices option).
-func storageRsyncCopy(source string, dest string) (string, error) {
-	err := os.MkdirAll(dest, 0755)
-	if err != nil {
-		return "", err
-	}
-
-	rsyncVerbosity := "-q"
-	if debug {
-		rsyncVerbosity = "-vi"
-	}
-
-	output, err := shared.RunCommand(
-		"rsync",
-		"-a",
-		"-HAX",
-		"--devices",
-		"--delete",
-		"--checksum",
-		"--numeric-ids",
-		rsyncVerbosity,
-		shared.AddSlash(source),
-		dest)
-
-	return output, err
-}
-
 // storageType defines the type of a storage
 type storageType int
 

lxd/storage_btrfs.go

@@ -463,7 +463,11 @@ func (s *storageBtrfs) StoragePoolUmount() (bool, error) {
 }
 
 func (s *storageBtrfs) StoragePoolUpdate(writable *api.StoragePoolPut, changedConfig []string) error {
-	return fmt.Errorf("Btrfs storage properties cannot be changed.")
+	if shared.StringInSlice("rsync.bwlimit", changedConfig) {
+		return nil
+	}
+
+	return fmt.Errorf("Storage property cannot be changed.")
 }
 
 func (s *storageBtrfs) GetStoragePoolWritable() api.StoragePoolPut {
@@ -1005,7 +1009,8 @@ func (s *storageBtrfs) ContainerRestore(container container, sourceContainer con
 		if err == nil {
 			// Use rsync to fill the empty volume.  Sync by using
 			// the subvolume name.
-			output, err := storageRsyncCopy(sourceContainerSubvolumeName, targetContainerSubvolumeName)
+			bwlimit := s.pool.Config["rsync.bwlimit"]
+			output, err := rsyncLocalCopy(sourceContainerSubvolumeName, targetContainerSubvolumeName, bwlimit)
 			if err != nil {
 				s.ContainerDelete(container)
 				logger.Errorf("ContainerRestore: rsync failed: %s.", string(output))
@@ -1743,7 +1748,7 @@ func (s *btrfsMigrationSourceDriver) send(conn *websocket.Conn, btrfsPath string
 	return err
 }
 
-func (s *btrfsMigrationSourceDriver) SendWhileRunning(conn *websocket.Conn, op *operation) error {
+func (s *btrfsMigrationSourceDriver) SendWhileRunning(conn *websocket.Conn, op *operation, bwlimit string) error {
 	_, containerPool := s.container.Storage().GetContainerPoolInfo()
 	containerName := s.container.Name()
 	containersPath := getContainerMountPoint(containerPool, "")
@@ -1821,7 +1826,7 @@ func (s *btrfsMigrationSourceDriver) SendWhileRunning(conn *websocket.Conn, op *
 	return s.send(conn, migrationSendSnapshot, btrfsParent, wrapper)
 }
 
-func (s *btrfsMigrationSourceDriver) SendAfterCheckpoint(conn *websocket.Conn) error {
+func (s *btrfsMigrationSourceDriver) SendAfterCheckpoint(conn *websocket.Conn, bwlimit string) error {
 	tmpPath := containerPath(fmt.Sprintf("%s/.migration-send", s.container.Name()), true)
 	err := os.MkdirAll(tmpPath, 0700)
 	if err != nil {

lxd/storage_dir.go

@@ -159,7 +159,11 @@ func (s *storageDir) GetContainerPoolInfo() (int64, string) {
 }
 
 func (s *storageDir) StoragePoolUpdate(writable *api.StoragePoolPut, changedConfig []string) error {
-	return fmt.Errorf("Dir storage properties cannot be changed.")
+	if shared.StringInSlice("rsync.bwlimit", changedConfig) {
+		return nil
+	}
+
+	return fmt.Errorf("Storage property cannot be changed.")
 }
 
 // Functions dealing with storage pools.
@@ -367,7 +371,8 @@ func (s *storageDir) copyContainer(target container, source container) error {
 		return err
 	}
 
-	output, err := storageRsyncCopy(sourceContainerMntPoint, targetContainerMntPoint)
+	bwlimit := s.pool.Config["rsync.bwlimit"]
+	output, err := rsyncLocalCopy(sourceContainerMntPoint, targetContainerMntPoint, bwlimit)
 	if err != nil {
 		return fmt.Errorf("Failed to rsync container: %s: %s.", string(output), err)
 	}
@@ -400,7 +405,8 @@ func (s *storageDir) copySnapshot(target container, source container) error {
 		return err
 	}
 
-	output, err := storageRsyncCopy(sourceContainerMntPoint, targetContainerMntPoint)
+	bwlimit := s.pool.Config["rsync.bwlimit"]
+	output, err := rsyncLocalCopy(sourceContainerMntPoint, targetContainerMntPoint, bwlimit)
 	if err != nil {
 		return fmt.Errorf("Failed to rsync container: %s: %s.", string(output), err)
 	}
@@ -533,7 +539,8 @@ func (s *storageDir) ContainerRestore(container container, sourceContainer conta
 	sourcePath := sourceContainer.Path()
 
 	// Restore using rsync
-	output, err := storageRsyncCopy(sourcePath, targetPath)
+	bwlimit := s.pool.Config["rsync.bwlimit"]
+	output, err := rsyncLocalCopy(sourcePath, targetPath, bwlimit)
 	if err != nil {
 		return fmt.Errorf("Failed to rsync container: %s: %s.", string(output), err)
 	}
@@ -566,8 +573,8 @@ func (s *storageDir) ContainerSnapshotCreate(snapshotContainer container, source
 		return err
 	}
 
-	rsync := func(snapshotContainer container, oldPath string, newPath string) error {
-		output, err := storageRsyncCopy(oldPath, newPath)
+	rsync := func(snapshotContainer container, oldPath string, newPath string, bwlimit string) error {
+		output, err := rsyncLocalCopy(oldPath, newPath, bwlimit)
 		if err != nil {
 			s.ContainerDelete(snapshotContainer)
 			return fmt.Errorf("Failed to rsync: %s: %s.", string(output), err)
@@ -586,7 +593,8 @@ func (s *storageDir) ContainerSnapshotCreate(snapshotContainer container, source
 	_, sourcePool := sourceContainer.Storage().GetContainerPoolInfo()
 	sourceContainerName := sourceContainer.Name()
 	sourceContainerMntPoint := getContainerMountPoint(sourcePool, sourceContainerName)
-	err = rsync(snapshotContainer, sourceContainerMntPoint, targetContainerMntPoint)
+	bwlimit := s.pool.Config["rsync.bwlimit"]
+	err = rsync(snapshotContainer, sourceContainerMntPoint, targetContainerMntPoint, bwlimit)
 	if err != nil {
 		return err
 	}
@@ -602,7 +610,7 @@ func (s *storageDir) ContainerSnapshotCreate(snapshotContainer container, source
 			return nil
 		}
 
-		err = rsync(snapshotContainer, sourceContainerMntPoint, targetContainerMntPoint)
+		err = rsync(snapshotContainer, sourceContainerMntPoint, targetContainerMntPoint, bwlimit)
 		if err != nil {
 			return err
 		}