X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/ba15fa5da21f4bafd3f90a8d259ea2aae764c77e..3194c1b24ffe6fff5fcb2f620ca6ee43741e3462:/services/crunch-run/crunchrun.go

diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go
index 6a118a6743..1bd49569d3 100644
--- a/services/crunch-run/crunchrun.go
+++ b/services/crunch-run/crunchrun.go
@@ -6,7 +6,6 @@ package main
 
 import (
 	"bytes"
-	"context"
 	"encoding/json"
 	"errors"
 	"flag"
@@ -33,6 +32,7 @@ import (
 	"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
 	"git.curoverse.com/arvados.git/sdk/go/keepclient"
 	"git.curoverse.com/arvados.git/sdk/go/manifest"
+	"golang.org/x/net/context"
 
 	dockertypes "github.com/docker/docker/api/types"
 	dockercontainer "github.com/docker/docker/api/types/container"
@@ -82,53 +82,6 @@ type ThinDockerClient interface {
 	ImageRemove(ctx context.Context, image string, options dockertypes.ImageRemoveOptions) ([]dockertypes.ImageDeleteResponseItem, error)
 }
 
-// ThinDockerClientProxy is a proxy implementation of ThinDockerClient
-// that executes the docker requests on dockerclient.Client
-type ThinDockerClientProxy struct {
-	Docker *dockerclient.Client
-}
-
-// ContainerAttach invokes dockerclient.Client.ContainerAttach
-func (proxy ThinDockerClientProxy) ContainerAttach(ctx context.Context, container string, options dockertypes.ContainerAttachOptions) (dockertypes.HijackedResponse, error) {
-	return proxy.Docker.ContainerAttach(ctx, container, options)
-}
-
-// ContainerCreate invokes dockerclient.Client.ContainerCreate
-func (proxy ThinDockerClientProxy) ContainerCreate(ctx context.Context, config *dockercontainer.Config, hostConfig *dockercontainer.HostConfig,
-	networkingConfig *dockernetwork.NetworkingConfig, containerName string) (dockercontainer.ContainerCreateCreatedBody, error) {
-	return proxy.Docker.ContainerCreate(ctx, config, hostConfig, networkingConfig, containerName)
-}
-
-// ContainerStart invokes dockerclient.Client.ContainerStart
-func (proxy ThinDockerClientProxy) ContainerStart(ctx context.Context, container string, options dockertypes.ContainerStartOptions) error {
-	return proxy.Docker.ContainerStart(ctx, container, options)
-}
-
-// ContainerStop invokes dockerclient.Client.ContainerStop
-func (proxy ThinDockerClientProxy) ContainerStop(ctx context.Context, container string, timeout *time.Duration) error {
-	return proxy.Docker.ContainerStop(ctx, container, timeout)
-}
-
-// ContainerWait invokes dockerclient.Client.ContainerWait
-func (proxy ThinDockerClientProxy) ContainerWait(ctx context.Context, container string, condition dockercontainer.WaitCondition) (<-chan dockercontainer.ContainerWaitOKBody, <-chan error) {
-	return proxy.Docker.ContainerWait(ctx, container, condition)
-}
-
-// ImageInspectWithRaw invokes dockerclient.Client.ImageInspectWithRaw
-func (proxy ThinDockerClientProxy) ImageInspectWithRaw(ctx context.Context, image string) (dockertypes.ImageInspect, []byte, error) {
-	return proxy.Docker.ImageInspectWithRaw(ctx, image)
-}
-
-// ImageLoad invokes dockerclient.Client.ImageLoad
-func (proxy ThinDockerClientProxy) ImageLoad(ctx context.Context, input io.Reader, quiet bool) (dockertypes.ImageLoadResponse, error) {
-	return proxy.Docker.ImageLoad(ctx, input, quiet)
-}
-
-// ImageRemove invokes dockerclient.Client.ImageRemove
-func (proxy ThinDockerClientProxy) ImageRemove(ctx context.Context, image string, options dockertypes.ImageRemoveOptions) ([]dockertypes.ImageDeleteResponseItem, error) {
-	return proxy.Docker.ImageRemove(ctx, image, options)
-}
-
 // ContainerRunner is the main stateful struct used for a single execution of a
 // container.
 type ContainerRunner struct {
@@ -1086,14 +1039,34 @@ func (runner *ContainerRunner) UploadOutputFile(
 	relocateTo string,
 	followed int) (manifestText string, err error) {
 
-	if info.Mode().IsDir() {
-		return
-	}
-
 	if infoerr != nil {
 		return "", infoerr
 	}
 
+	if info.Mode().IsDir() {
+		// if empty, need to create a .keep file
+		dir, direrr := os.Open(path)
+		if direrr != nil {
+			return "", direrr
+		}
+		defer dir.Close()
+		names, eof := dir.Readdirnames(1)
+		if len(names) == 0 && eof == io.EOF && path != runner.HostOutputDir {
+			containerPath := runner.OutputPath + path[len(runner.HostOutputDir):]
+			for _, bind := range binds {
+				mnt := runner.Container.Mounts[bind]
+				// Check if there is a bind for this
+				// directory, in which case assume we don't need .keep
+				if (containerPath == bind || strings.HasPrefix(containerPath, bind+"/")) && mnt.PortableDataHash != "d41d8cd98f00b204e9800998ecf8427e+0" {
+					return
+				}
+			}
+			outputSuffix := path[len(runner.HostOutputDir)+1:]
+			return fmt.Sprintf("./%v d41d8cd98f00b204e9800998ecf8427e+0 0:0:.keep\n", outputSuffix), nil
+		}
+		return
+	}
+
 	if followed >= limitFollowSymlinks {
 		// Got stuck in a loop or just a pathological number of
 		// directory links, give up.
@@ -1101,9 +1074,16 @@ func (runner *ContainerRunner) UploadOutputFile(
 		return
 	}
 
-	// When following symlinks, the source path may need to be logically
-	// relocated to some other path within the output collection.  Remove
-	// the relocateFrom prefix and replace it with relocateTo.
+	// "path" is the actual path we are visiting
+	// "tgt" is the target of "path" (a non-symlink) after following symlinks
+	// "relocated" is the path in the output manifest where the file should be placed,
+	// but has HostOutputDir as a prefix.
+
+	// The destination path in the output manifest may need to be
+	// logically relocated to some other path in order to appear
+	// in the correct location as a result of following a symlink.
+	// Remove the relocateFrom prefix and replace it with
+	// relocateTo.
 	relocated := relocateTo + path[len(relocateFrom):]
 
 	tgt, readlinktgt, info, derefErr := runner.derefOutputSymlink(path, info)
@@ -1124,7 +1104,7 @@ func (runner *ContainerRunner) UploadOutputFile(
 
 			// Terminates in this keep mount, so add the
 			// manifest text at appropriate location.
-			outputSuffix := path[len(runner.HostOutputDir):]
+			outputSuffix := relocated[len(runner.HostOutputDir):]
 			manifestText, err = runner.getCollectionManifestForPath(adjustedMount, outputSuffix)
 			return
 		}
@@ -1712,10 +1692,8 @@ func main() {
 	// API version 1.21 corresponds to Docker 1.9, which is currently the
 	// minimum version we want to support.
 	docker, dockererr := dockerclient.NewClient(dockerclient.DefaultDockerHost, "1.21", nil, nil)
-	dockerClientProxy := ThinDockerClientProxy{Docker: docker}
-
-	cr := NewContainerRunner(api, kc, dockerClientProxy, containerId)
 
+	cr := NewContainerRunner(api, kc, docker, containerId)
 	if dockererr != nil {
 		cr.CrunchLog.Printf("%s: %v", containerId, dockererr)
 		cr.checkBrokenNode(dockererr)