Bugfixes for the OSDF staging code.

With this, end-to-end tests of the staging binary works for a simple
job ad.

Author: bbockelm

cmd/osdf_stage/main.go

@@ -44,7 +44,7 @@ type Options struct {
 	OriginPrefix string `long:"origin-prefix" short:"o" description:"Prefix corresponding to the local origin"`
 
 	// Shadow origin prefix; e.g., osdf://ospool/osgconnect-shadow/
-	ShadowOriginPrefix string `long:"shadow-prefix" short:"s" description:"Prefix corresponding to the shadow origin"`
+	ShadowOriginPrefix string `long:"shadow-prefix" short:"s" description:"Prefix corresponding to the shadow origin" required:"true"`
 
 	// Sources to ingest
 	Sources []string `positional-arg-name:"sources" short:"i" long:"input" description:"Source file(s)" default:"-"`
@@ -59,6 +59,22 @@ func main() {
 	// Capture the start time of the transfer
 	if _, err := parser.Parse(); err != nil {
 		if flagsErr, ok := err.(*flags.Error); ok && flagsErr.Type == flags.ErrHelp {
+			fmt.Fprintln(os.Stderr, `
+This utility parses a job ClassAd and, for each "osdf://" URL found in
+the input files that is in a locally-mounted origin, copies the file
+over to a "shadow origin".  The files in the shadow origin are given a
+unique based on their last modification time; this means that local
+files can be modified without causing cache consistency issues.
+
+Terminology:
+- Origin prefix: Where in the OSDF namespace the origin exports its
+  files.  Example: osdf://osg-connect/protected
+- Mount prefix: The location in the locally-mounted filesystem that
+  correspondings to the files in the origin prefix. Example:
+  /mnt/cephfs/protected
+- Shadow prefix: Where in the OSDF namespace the resulting files should
+  be uploaded.  Example: osdf://osg-connect-shadow/protected
+`);
 			os.Exit(0)
 		} else {
 			log.Errorln(err)
@@ -89,7 +105,8 @@ func main() {
 		log.Errorln("Origin prefix scheme must be osdf://:", originPrefixUri.Scheme)
 		os.Exit(1)
 	}
-	originPrefixPath := path.Clean("/" + originPrefixUri.Path)
+	originPrefixPath := path.Clean("/" + originPrefixUri.Host + "/" + originPrefixUri.Path)
+	log.Debugln("Local origin prefix:", originPrefixPath)
 
 	if options.Version {
 		fmt.Println("Version:", version)
@@ -126,7 +143,7 @@ func main() {
 			os.Exit(1)
 		}
 		inputList, err := classad.Get("TransferInput")
-		if err != nil {
+		if err != nil || inputList == nil {
 			// No TransferInput, no need to transform...
 			os.Exit(0)
 		}
@@ -137,14 +154,15 @@ func main() {
 		}
 		re := regexp.MustCompile("[,\\s]+")
 		for _, source := range re.Split(inputListStr, -1) {
+			log.Debugln("Examining transfer input file", source)
 			if (strings.HasPrefix(source, options.MountPrefix)) {
 				sources = append(sources, source)
 			} else {
 					// Replace the osdf:// prefix with the local mount path
 				source_uri, err := url.Parse(source)
 				source_uri_scheme := strings.SplitN(source_uri.Scheme, "+", 2)[0]
-				if err != nil && source_uri_scheme == "osdf" {
-					source_path := path.Clean("/" + source_uri.Path)
+				if err == nil && source_uri_scheme == "osdf" {
+					source_path := path.Clean("/" + source_uri.Host + "/" + source_uri.Path)
 					if (strings.HasPrefix(source_path, originPrefixPath)) {
 						sources = append(sources, options.MountPrefix + source_path[len(originPrefixPath):])
 						continue
@@ -169,7 +187,14 @@ func main() {
 	for _, src := range sources {
 		_, newSource, result := stashcp.DoShadowIngest(src, options.MountPrefix, options.ShadowOriginPrefix)
 		if result != nil {
-			break
+			// What's the correct behavior on failure?  For now, we silently put the transfer
+			// back on the original list.  This is arguably the wrong approach as it might
+			// give the user surprising semantics -- but keeping this until we have a bit more
+			// confidence in the approach.
+			extraSources = append(extraSources, src)
+			log.Errorf("Failed to ingest %s: %s.  Adding original back to the transfer list",
+				src, result.Error())
+			continue
 		}
 		xformSources = append(xformSources, newSource)
 	}

handle_http.go

@@ -39,6 +39,16 @@ func (e *StoppedTransferError) Error() string {
 }
 
 
+type HttpErrResp struct {
+	Code int
+	Err string
+}
+
+func (e *HttpErrResp) Error() string {
+	return e.Err
+}
+
+
 // SlowTransferError is an error that is returned when a transfer takes longer than the configured timeout
 type SlowTransferError struct {
 	BytesTransferred int64
@@ -898,9 +908,12 @@ func walkDir(path string, client *gowebdav.Client) ([]string, error) {
 	return files, nil
 }
 
-func stat_http(dest *url.URL, namespace Namespace) (uint64, error) {
+func StatHttp(dest *url.URL, namespace Namespace) (uint64, error) {
 
-	token_name := getTokenName(dest)
+	scitoken_contents, err := getToken(dest, namespace, false, "")
+	if err != nil {
+		return 0, err
+	}
 
 	// Parse the writeback host as a URL
 	writebackhostUrl, err := url.Parse(namespace.WriteBackHost)
@@ -910,19 +923,9 @@ func stat_http(dest *url.URL, namespace Namespace) (uint64, error) {
 	dest.Host = writebackhostUrl.Host
 	dest.Scheme = "https"
 
-	var token string
-	if namespace.UseTokenOnRead {
-		token, err = getToken(token_name)
-		if err != nil {
-			log.Errorln("Failed to get token though required to read from this namespace:", err)
-			return 0, err
-		}
-	}
-
-	_, canDisableProxy := os.LookupEnv("OSG_DISABLE_PROXY_FALLBACK")
-	canDisableProxy = !canDisableProxy
+	canDisableProxy := CanDisableProxy()
+	disableProxy := !IsProxyEnabled()
 
-	disableProxy := false
 	var resp *http.Response
 	for {
 		defaultTransport := http.DefaultTransport.(*http.Transport).Clone()
@@ -940,8 +943,8 @@ func stat_http(dest *url.URL, namespace Namespace) (uint64, error) {
 			return 0, err
 		}
 
-		if token != "" {
-			req.Header.Set("Authorization", "Bearer " + token)
+		if scitoken_contents != "" {
+			req.Header.Set("Authorization", "Bearer " + scitoken_contents)
 		}
 
 		resp, err = client.Do(req)
@@ -979,7 +982,7 @@ func stat_http(dest *url.URL, namespace Namespace) (uint64, error) {
 			return 0, err
 		}
 		defer resp.Body.Close()
-		return 0, errors.New(fmt.Sprintf("Request failed (HTTP status %d): %s", resp.StatusCode, string(response_b)))
+		return 0, &HttpErrResp{resp.StatusCode, fmt.Sprintf("Request failed (HTTP status %d): %s", resp.StatusCode, string(response_b))}
 	}
 }
 

handle_ingest.go

@@ -37,6 +37,7 @@ func generate_destination(filePath string, originPrefix string, shadowOriginPref
 func DoShadowIngest(sourceFile string, originPrefix string, shadowOriginPrefix string) (int64, string, error) {
 	for idx := 0; idx < 10; idx++ {
 		shadowFile, localSize, err := generate_destination(sourceFile, originPrefix, shadowOriginPrefix)
+		log.Debugln("Resulting shadow URL:", shadowFile)
 		if err != nil {
 			return 0, "", err
 		}
@@ -48,12 +49,19 @@ func DoShadowIngest(sourceFile string, originPrefix string, shadowOriginPrefix s
 		maxRuntime := float64(localSize / 10*1024*1024) + 300
 		for {
 			remoteSize, err := CheckOSDF(shadowFile, methods)
-			if err != nil {
+			if httpErr, ok := err.(*HttpErrResp); ok {
+				if httpErr.Code == 404 {
+					break
+				} else {
+					return 0, "", err
+				}
+			} else if err != nil {
 				return 0, "", err
 			}
 			if (localSize == remoteSize) {
 				return 0, shadowFile, err
 			}
+			log.Debugf("Remote file exists but it is incorrect size; actual size %v, expected %v.", remoteSize, localSize)
 
 			// If the remote file size is growing, then wait a bit; perhaps someone
 			// else is uploading the same file concurrently.
@@ -71,8 +79,9 @@ func DoShadowIngest(sourceFile string, originPrefix string, shadowOriginPrefix s
 				log.Warnln("Remote uploader took too long to upload file; will do upload from this client")
 				break
 			}
+			log.Debugf("Will sleep for 5 seconds to see if another client is currently uploading the file")
 			// TODO: Could use a clever backoff scheme here.
-			time.Sleep(5)
+			time.Sleep(5 * time.Second)
 		}
 
 		uploadBytes, err := DoStashCPSingle(sourceFile, shadowFile, methods, false)

main.go

@@ -227,6 +227,10 @@ func CheckOSDF(destination string, methods []string) (remoteSize uint64, err err
 	if dest_uri.Scheme == "" {
 		dest_uri.Scheme = "osdf"
 	}
+	if dest_uri.Host != "" {
+		dest_uri.Path = path.Clean("/" + dest_uri.Host + "/" + dest_uri.Path)
+		dest_uri.Host = ""
+	}
 
 	ns, err := MatchNamespace(dest_uri.Path)
 	if err != nil {
@@ -238,7 +242,7 @@ func CheckOSDF(destination string, methods []string) (remoteSize uint64, err err
 		switch method {
 		case "http":
 			log.Info("Trying HTTP...")
-			if remoteSize, err = stat_http(dest_uri, ns); err == nil {
+			if remoteSize, err = StatHttp(dest_uri, ns); err == nil {
 				return remoteSize, nil
 			}
 		default:
@@ -339,7 +343,7 @@ func DoStashCPSingle(sourceFile string, destination string, methods []string, re
 	}
 
 	if dest_url.Host != "" {
-		dest_url.Path = path.Join(dest_url.Host, dest_url.Path)
+		dest_url.Path = "/" + path.Join(dest_url.Host, dest_url.Path)
 	}
 
 	sourceScheme, _ := getTokenName(source_url)