Working Swift object storage access for worker

src/ga4gh-engine/worker/container_engine.go

@@ -31,7 +31,7 @@ func (self *DockerEngine) Run(containerName string, args []string,
 	list, err := self.client.ImageList(context.Background(), types.ImageListOptions{MatchName: containerName})
 
 	if err != nil || len(list) == 0 {
-		log.Printf("Image %s not found", containerName)
+		log.Printf("Image %s not found: %s", containerName, err)
 		pull_opt := types.ImagePullOptions{}
 		r, err := self.client.ImagePull(context.Background(), containerName, pull_opt)
 		if err != nil {

src/ga4gh-engine/worker/file_client.go

@@ -84,7 +84,10 @@ func (self *FileMapper) MapInput(jobId string, storage string, mountPath string,
 		if len(base) > 0 {
 			dstPath := path.Join(vol.HostPath, relpath)
 			fmt.Printf("get %s %s\n", storage, dstPath)
-			self.fileSystem.Get(storage, dstPath)
+			err := self.fileSystem.Get(storage, dstPath)
+			if err != nil {
+				return err
+			}
 		}
 	}
 	return nil

src/ga4gh-engine/worker/swift_client.go

@@ -5,13 +5,15 @@ import (
 	"github.com/rackspace/gophercloud/openstack"
 	"io"
 	"os"
+	"fmt"
 	"strings"
+	"log"
 	//"github.com/rackspace/gophercloud/openstack/objectstorage/v1/containers"
 	"github.com/rackspace/gophercloud/openstack/objectstorage/v1/objects"
 	//"github.com/rackspace/gophercloud/pagination"
 )
 
-var SWIFT_PROTOCOL = "sw://"
+var SWIFT_PROTOCOL = "swift://"
 
 type SwiftAccess struct {
 	client *gophercloud.ServiceClient
@@ -36,40 +38,49 @@ func NewSwiftAccess() *SwiftAccess {
 }
 
 func (self *SwiftAccess) Get(storage string, hostPath string) error {
+	log.Printf("Starting download of %s", storage)
 	storage = strings.TrimPrefix(storage, SWIFT_PROTOCOL)
-
-	opts := objects.DownloadOpts{}
+	storage_split := strings.SplitN(storage, "/", 2)
 
 	// Download everything into a DownloadResult struct
-	res := objects.Download(self.client, "container_name", "object_name", opts)
+	opts := objects.DownloadOpts{}
+	res := objects.Download(self.client, storage_split[0], storage_split[1], opts)
 
 	file, err := os.Create(hostPath)
 	if err != nil {
 		return err
 	}
 	buffer := make([]byte, 10240)
+	total_len := 0
 	for {
 		len, err := res.Body.Read(buffer)
 		if err == io.EOF {
 			break
+		} else if err != nil {
+			return fmt.Errorf("Error reading file")
 		}
+		total_len += len
 		file.Write(buffer[:len])
 	}
 	file.Close()
 	res.Body.Close()
+	log.Printf("Downloaded %d bytes", total_len)
 	return nil
 }
 
-func (self *SwiftAccess) Put(location string, hostPath string, directory bool) error {
-	opts := objects.CreateOpts{}
-
+func (self *SwiftAccess) Put(storage string, hostPath string, directory bool) error {
+	log.Printf("Starting upload of %s", storage)
 	content, err := os.Open(hostPath)
 	if err != nil {
 		return err
 	}
+
+	storage = strings.TrimPrefix(storage, SWIFT_PROTOCOL)
+	storage_split := strings.SplitN(storage, "/", 2)
 
 	// Now execute the upload
-	res := objects.Create(self.client, "container_name", "object_name", content, opts)
+	opts := objects.CreateOpts{}
+	res := objects.Create(self.client, storage_split[0], storage_split[1], content, opts)
 	_, err = res.ExtractHeader()
 	content.Close()
 	return err