Fixing output files not staging issues when wildcards are specified
@@ -57,6 +57,7 @@ import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -163,41 +164,48 @@ public class SCPDataStageTask implements Task {
if (fileName.contains("*")) {
String destParentPath = (new File(destinationURI.getPath())).getParentFile().getPath();
String sourceParentPath = (new File(sourceURI.getPath())).getParentFile().getPath();
+
+ log.info("Fetching output files for wildcard " + fileName + " in path " + sourceParentPath);
List<String> fileNames = taskContext.getParentProcessContext().getDataMovementRemoteCluster()
.getFileNameFromExtension(fileName, sourceParentPath, remoteSession);
ExperimentCatalog experimentCatalog = processContext.getExperimentCatalog();
+ log.info("File names that matched with wildcard " + fileName + " : " + fileNames.toString());
+
String experimentId = processContext.getExperimentId();
String processId = processContext.getProcessId();
OutputDataObjectType processOutput = taskContext.getProcessOutput();
- for(int i=0; i<fileNames.size(); i++){
- String temp = fileNames.get(i);
- if(temp != null && temp != ""){
+ for (String temp : fileNames) {
+ if (temp != null && !"".equals(temp)) {
fileName = temp;
}
+
if (destParentPath.endsWith(File.separator)) {
destinationURI = new URI(destParentPath + fileName);
} else {
destinationURI = new URI(destParentPath + File.separator + fileName);
}
-
//Wildcard support is only enabled for output data staging
if (processState == ProcessState.OUTPUT_DATA_STAGING) {
+ URI newSourceURI = new URI((sourceParentPath.endsWith(File.separator) ?
+ sourceParentPath : sourceParentPath + File.separator) +
+ fileName);
processOutput.setName(fileName);
- experimentCatalog.add(ExpCatChildDataType.EXPERIMENT_OUTPUT, Arrays.asList(processOutput), experimentId);
- experimentCatalog.add(ExpCatChildDataType.PROCESS_OUTPUT, Arrays.asList(processOutput), processId);
+ experimentCatalog.add(ExpCatChildDataType.EXPERIMENT_OUTPUT, Collections.singletonList(processOutput), experimentId);
+ experimentCatalog.add(ExpCatChildDataType.PROCESS_OUTPUT, Collections.singletonList(processOutput), processId);
taskContext.setProcessOutput(processOutput);
makeDir(taskContext, destinationURI);
// TODO - save updated subtask model with new destination
- outputDataStaging(taskContext, remoteSession, sourceURI, storageSession, destinationURI);
- status.setReason("Successfully staged output data");
+ log.info("Staging derived output file " + fileName + " from " + newSourceURI.toString());
+ outputDataStaging(taskContext, remoteSession, newSourceURI, storageSession, destinationURI);
+ status.setReason("Successfully staged output file " + fileName);
}
}
if (processState == ProcessState.OUTPUT_DATA_STAGING) {
@@ -309,11 +317,13 @@ public class SCPDataStageTask implements Task {
*/
//Wildcard file path has not been resolved and cannot be handled. Hence ignoring
if(!destinationURI.toString().contains("*")) {
+ log.info("Starting file transfer from " + sourceURI.toString() + " to " + destinationURI.toString());
taskContext.getParentProcessContext().getDataMovementRemoteCluster().scpThirdParty(sourceURI.getPath(), srcSession,
destinationURI.getPath(), destSession, RemoteCluster.DIRECTION.TO, true);
// update output locations
GFacUtils.saveExperimentOutput(taskContext.getParentProcessContext(), taskContext.getProcessOutput().getName(), destinationURI.toString());
GFacUtils.saveProcessOutput(taskContext.getParentProcessContext(), taskContext.getProcessOutput().getName(), destinationURI.toString());
+ log.info("Finished file transfer to " + destinationURI.toString());
} else {
log.warn("Destination file path contains unresolved wildcards. Path: " + destinationURI.toString());
}
