Merge pull request #41 from RSE-Cambridge/rsync-fixes

Rsync fixes

Author: JohnGarbutt

dac-ansible/roles/data-acc/tasks/main.yml

@@ -1,13 +1,5 @@
 ---
-- name: check for installation of data-acc
-  become: yes
-  stat:
-    path: '{{data_acc_install_dir}}/{{data_acc_name}}/bin'
-  changed_when: false
-  register: data_acc_binary_dir
-
-- when: not data_acc_binary_dir.stat.exists
-  block:
+- block:
     - name: download data_acc
       become: yes
       become_user: root
@@ -33,7 +25,6 @@
         remote_src: yes
         src: /tmp/{{data_acc_tgz}}
         dest: '{{data_acc_install_dir}}/{{data_acc_name}}'
-        creates: '{{data_acc_install_dir}}/{{data_acc_name}}/bin'
   always:
     - name: delete archive
       become: yes

docker-slurm/burst_buffer.conf

@@ -1,4 +1,3 @@
-AllowUsers=root,slurm
 Flags=EnablePersistent,PrivateData
 
 StageInTimeout=3600

internal/pkg/dacctl/persistent.go

@@ -117,8 +117,6 @@ func CreateVolumesAndJobs(volReg registry.VolumeRegistry, poolRegistry registry.
 		JobVolume: volume.Name, // Even though its a persistent buffer, we add it here to ensure we delete buffer
 		Paths:     make(map[string]string),
 	}
-	job.Paths[fmt.Sprintf("DW_PERSISTENT_STRIPED_%s", volume.Name)] = fmt.Sprintf(
-		"/mnt/dac/job/%s/multijob/%s", job.Name, volume.Name)
 
 	err = volReg.AddJob(job)
 	if err != nil {

internal/pkg/pfsprovider/ansible/copy.go

@@ -4,16 +4,36 @@ import (
 	"fmt"
 	"github.com/RSE-Cambridge/data-acc/internal/pkg/registry"
 	"log"
+	"path"
+	"strings"
 )
 
 func processDataCopy(volume registry.Volume, request registry.DataCopyRequest) error {
 	cmd, err := generateDataCopyCmd(volume, request)
 	if err != nil {
 		return err
 	}
+	if cmd == "" {
+		log.Println("No files to copy for:", volume.Name)
+		return nil
+	}
+
+	log.Printf("Doing copy: %s", cmd)
 
-	log.Printf("FAKE copy: %s", cmd)
-	return nil
+	// Make sure global dir is setup correctly
+	// TODO: share code with mount better
+	// TODO: Probably should all get setup in fs-ansible really!!
+	mountDir := fmt.Sprintf("/mnt/lustre/%s", volume.UUID)
+	sharedDir := path.Join(mountDir, "/global")
+	if err := mkdir("localhost", sharedDir); err != nil {
+		return err
+	}
+	if err := fixUpOwnership("localhost", volume.Owner, volume.Group, sharedDir); err != nil {
+		return err
+	}
+
+	// Do the copy
+	return runner.Execute("localhost", cmd)
 }
 
 func generateDataCopyCmd(volume registry.Volume, request registry.DataCopyRequest) (string, error) {
@@ -22,24 +42,31 @@ func generateDataCopyCmd(volume registry.Volume, request registry.DataCopyReques
 		return "", err
 	}
 
-	cmd := fmt.Sprintf("sudo su `getent passwd %d | cut -d: -f1` %s", volume.Owner, rsync)
+	cmd := fmt.Sprintf("sudo -g '#%d' -u '#%d' %s", volume.Group, volume.Owner, rsync)
+	dacHostBufferPath := fmt.Sprintf("/mnt/lustre/%s/global", volume.UUID)
+	cmd = fmt.Sprintf("bash -c \"export DW_JOB_STRIPED='%s' && %s\"", dacHostBufferPath, cmd)
 	return cmd, nil
 }
 
 func generateRsyncCmd(volume registry.Volume, request registry.DataCopyRequest) (string, error) {
 	if request.Source == "" && request.Destination == "" {
-		log.Println("No files to copy for:", volume.Name)
 		return "", nil
 	}
 
 	var flags string
 	if request.SourceType == registry.Directory {
-		flags = "-r "
+		flags = "-r -ospgu --stats"
 	} else if request.SourceType == registry.File {
-		flags = ""
+		flags = "-ospgu --stats"
 	} else {
 		return "", fmt.Errorf("unsupported source type %s for volume: %s", request.SourceType, volume.Name)
 	}
 
-	return fmt.Sprintf("rsync %s%s %s", flags, request.Source, request.Destination), nil
+	return fmt.Sprintf("rsync %s %s %s", flags,
+		escapePath(request.Source),
+		escapePath(request.Destination)), nil
+}
+
+func escapePath(path string) string {
+	return strings.Replace(path, "$DW_JOB_STRIPED", "\\$DW_JOB_STRIPED", 1)
 }

internal/pkg/pfsprovider/ansible/copy_test.go

@@ -10,6 +10,8 @@ func Test_GenerateDataCopy(t *testing.T) {
 	testVolume := registry.Volume{
 		Name:  registry.VolumeName("asdf"),
 		Owner: 1001,
+		Group: 1002,
+		UUID:  "fsuuid",
 	}
 	request := registry.DataCopyRequest{}
 
@@ -18,11 +20,11 @@ func Test_GenerateDataCopy(t *testing.T) {
 	assert.Empty(t, cmd)
 
 	request.SourceType = registry.File
-	request.Source = "source"
+	request.Source = "$DW_JOB_STRIPED/source"
 	request.Destination = "dest"
 	cmd, err = generateDataCopyCmd(testVolume, request)
 	assert.Nil(t, err)
-	assert.Equal(t, "sudo su `getent passwd 1001 | cut -d: -f1` rsync source dest", cmd)
+	assert.Equal(t, "bash -c \"export DW_JOB_STRIPED='/mnt/lustre/fsuuid/global' && sudo -g '#1002' -u '#1001' rsync -ospgu --stats \\$DW_JOB_STRIPED/source dest\"", cmd)
 
 	request.SourceType = registry.List
 	request.Source = "list_filename"
@@ -47,14 +49,14 @@ func Test_GenerateRsyncCmd(t *testing.T) {
 	request.Destination = "dest"
 	cmd, err = generateRsyncCmd(testVolume, request)
 	assert.Nil(t, err)
-	assert.Equal(t, "rsync source dest", cmd)
+	assert.Equal(t, "rsync -ospgu --stats source dest", cmd)
 
 	request.SourceType = registry.Directory
 	request.Source = "source"
 	request.Destination = "dest"
 	cmd, err = generateRsyncCmd(testVolume, request)
 	assert.Nil(t, err)
-	assert.Equal(t, "rsync -r source dest", cmd)
+	assert.Equal(t, "rsync -r -ospgu --stats source dest", cmd)
 
 	request.SourceType = registry.List
 	request.Source = "list_filename"

tools/dac-reset.sh

@@ -7,8 +7,8 @@ set +a
 /usr/local/bin/etcdctl --key /etc/data-acc/pki/`hostname`.dac.hpc.cam.ac.uk-key.pem --cert /etc/data-acc/pki/`hostname`.dac.hpc.cam.ac.uk.pem --cacert /etc/data-acc/pki/ca.pem del --prefix ''
 
 # Kill all lustre filesystems
-ssh slurm-cpu1 sudo umount -atl lustre
-ssh slurm-cpu2 sudo umount -atl lustre
+#ssh slurm-cpu1 sudo umount -atl lustre
+#ssh slurm-cpu2 sudo umount -atl lustre
 ssh dac1 sudo umount -at lustre
 ssh dac2 sudo umount -at lustre
 ssh dac3 sudo umount -at lustre

tools/slurm-test.sh

@@ -12,11 +12,17 @@ echo "#!/bin/bash
 #DW jobdw capacity=2TB access_mode=striped,private type=scratch
 #DW persistentdw name=mytestbuffer
 #DW swap 5MB
-#DW stage_in source=/global/cscratch1/filename1 destination=\$DW_JOB_STRIPED/filename1 type=file
-#DW stage_out source=\$DW_JOB_STRIPED/outdir destination=/global/scratch1/outdir type=directory
+#DW stage_in source=/usr/local/bin/dacd destination=\$DW_JOB_STRIPED/filename1 type=file
+#DW stage_out source=\$DW_JOB_STRIPED/outdir destination=/tmp type=directory
+
 env
 df -h
 swapon
+
+mkdir \$DW_JOB_STRIPED/outdir
+df -h > \$DW_JOB_STRIPED/outdir/dfoutput
+ls -al \$DW_JOB_STRIPED > \$DW_JOB_STRIPED/outdir/lsoutput
+
 echo \$HOSTNAME
 " > use-persistent.sh
 
@@ -45,11 +51,14 @@ scontrol show burstbuffer
 squeue
 
 echo "***Use persistent buffer***"
+adduser centos
 cat use-persistent.sh
-su slurm -c 'sbatch use-persistent.sh'
-su slurm -c 'sbatch use-persistent.sh'
-su slurm -c 'sbatch use-persistent.sh'
-su slurm -c 'sbatch use-persistent.sh'
+su centos -c 'sbatch use-persistent.sh'
+su centos -c 'sbatch use-persistent.sh'
+su centos -c 'sbatch use-persistent.sh'
+su centos -c 'sbatch use-persistent.sh'
+su centos -c 'sbatch --array=1-10 test-persistent.sh'
+
 squeue
 
 sleep $SLEEP_INTERVAL