merge fixes from 0.10

src/python/dm/common/constants/dmProcessingStatus.py

@@ -5,6 +5,7 @@
 DM_PROCESSING_STATUS_ANY = 'any'
 DM_PROCESSING_STATUS_PENDING = 'pending'
 DM_PROCESSING_STATUS_RUNNING = 'running'
+DM_PROCESSING_STATUS_FINALIZING = 'finalizing'
 DM_PROCESSING_STATUS_DONE = 'done'
 DM_PROCESSING_STATUS_FAILED = 'failed'
 DM_PROCESSING_STATUS_SKIPPED = 'skipped'
@@ -15,6 +16,7 @@ DM_ALLOWED_PROCESSING_STATUS_LIST = [
     DM_PROCESSING_STATUS_ANY,
     DM_PROCESSING_STATUS_PENDING,
     DM_PROCESSING_STATUS_RUNNING,
+    DM_PROCESSING_STATUS_FINALIZING,
     DM_PROCESSING_STATUS_DONE,
     DM_PROCESSING_STATUS_FAILED,
     DM_PROCESSING_STATUS_SKIPPED,

src/python/dm/common/objects/daqInfo.py

@@ -76,10 +76,12 @@ class DaqInfo(DmObject):
         self['percentageProcessed'] = '%.2f' % percentageProcessed
         self['percentageProcessingErrors'] = '%.2f' % percentageProcessingErrors
 
-        if self.get('endTime') and nCompletedFiles == nFiles:
-            daqStatus = 'done'
-            if nProcessingErrors:
-                daqStatus = 'failed'
+        if self.get('endTime'):
+            daqStatus = dmProcessingStatus.DM_PROCESSING_STATUS_FINALIZING
+            if nCompletedFiles == nFiles:
+                daqStatus = dmProcessingStatus.DM_PROCESSING_STATUS_DONE
+                if nProcessingErrors:
+                    daqStatus = dmProcessingStatus.DM_PROCESSING_STATUS_FAILED
             lastFileProcessingErrorTime = self.get('lastFileProcessingErrorTime')
             lastFileProcessedTime = self.get('lastFileProcessedTime')
             endTime = lastFileProcessedTime

src/python/dm/common/processing/fileProcessingThread.py

@@ -2,6 +2,7 @@
 
 import threading
 import time
+from dm.common.constants import dmProcessingStatus
 
 from dm.common.utility.loggingManager import LoggingManager
 
@@ -31,7 +32,7 @@ class FileProcessingThread(threading.Thread):
 
         try:
             statusMonitor = fileInfo.get('statusMonitor')
-            if statusMonitor and statusMonitor.get('status') == 'aborting':
+            if statusMonitor and statusMonitor.get('status') ==  dmProcessingStatus.DM_PROCESSING_STATUS_ABORTING:
                 self.logger.debug('File %s processing is cancelled' % (filePath))
                 endProcessingTime = time.time() 
                 statusMonitor.fileProcessingCancelled(filePath, endProcessingTime) 
@@ -87,7 +88,7 @@ class FileProcessingThread(threading.Thread):
                         return 
                     else:
                         retryWaitPeriod = processor.getRetryWaitPeriodInSeconds()
-                        self.logger.debug('%s will retry processing file %s in %s seconds' % (processorName, filePath, retryWaitPeriod))
+                        self.logger.debug('%s may retry processing file %s after at least %s seconds' % (processorName, filePath, retryWaitPeriod))
                         self.fileProcessingQueue.push(fileInfo, retryWaitPeriod)
                         # Do not process this file further until 
                         # this plugin is done

src/python/dm/common/processing/plugins/fileTransferPlugin.py

@@ -34,15 +34,15 @@ class FileTransferPlugin(FileProcessor):
         self.start(src=srcUrl, dest=destUrl, templateInfo=fileInfo)
 
     def getSrcUrl(self, filePath, dataDirectory):
-        # Use relative path with respect to data directory as a source
-        srcUrl = os.path.relpath(filePath, dataDirectory)
-        return srcUrl
+        return filePath
     
     def getDestUrl(self, filePath, dataDirectory, storageHost, storageDirectory):
+        # Use relative path with respect to data directory as a source
+        srcUrl = os.path.relpath(filePath, dataDirectory)
         if self.dest:
-            destUrl = '%s' % (self.dest)
+            destUrl = '%s/%s' % (self.dest, srcUrl)
         else:
-            destUrl = '%s:%s' % (storageHost, storageDirectory)
+            destUrl = '%s:%s/%s' % (storageHost, storageDirectory, srcUrl)
         return destUrl
     
     def getSrcDirUrl(self, dataDirectory):

src/python/dm/common/processing/plugins/rsyncFileTransferPlugin.py

@@ -11,9 +11,9 @@ from dm.ds_web_service.api.dsRestApiFactory import DsRestApiFactory
 
 class RsyncFileTransferPlugin(FileTransferPlugin):
 
-    DEFAULT_COMMAND = 'rsync -arvlPR'
-    DIRECTORY_TRANSFER_COMMAND = 'rsync -arvlP'
-    DRY_RUN_COMMAND = 'rsync -arvlP'
+    DEFAULT_COMMAND = 'rsync -arvlPR --'
+    DIRECTORY_TRANSFER_COMMAND = 'rsync -arvlP --'
+    DRY_RUN_COMMAND = 'rsync -arvlP --dry-run --'
 
     def __init__(self, src=None, dest=None, command=DEFAULT_COMMAND, localMd5Sum=True, remoteMd5Sum=False, deleteOriginal=False, pluginMustProcessFiles=True, dependsOn=[]):
         FileTransferPlugin.__init__(self, command, src, dest, dependsOn=dependsOn)
@@ -29,7 +29,7 @@ class RsyncFileTransferPlugin(FileTransferPlugin):
         storageDirectory = uploadInfo['storageDirectory']
         storageHost = uploadInfo['storageHost']
         dataDirectory = uploadInfo['dataDirectory']
-        dryRunCommand = '%s --dry-run %s/ %s:%s' % (self.DRY_RUN_COMMAND, dataDirectory, storageHost, storageDirectory)
+        dryRunCommand = '%s %s/ %s:%s' % (self.DRY_RUN_COMMAND, dataDirectory, storageHost, storageDirectory)
         subprocess = DmSubprocess.getSubprocess(dryRunCommand)
         subprocess.run()
         lines = subprocess.getStdOut().split('\n')