Repository: dmwm/wmcore
Issue: phedex injector 400 error

We are seeing errors like this in the phedex injector:

2013-06-09 11:14:44,399:ERROR:PhEDExInjectorPoller:Encountered PassableError in PhEDExInjector
2013-06-09 11:14:44,399:ERROR:PhEDExInjectorPoller:Rolling back current transaction and terminating current loop, but not killing component.
2013-06-09 11:16:28,553:WARNING:Service:The cachefile /storage/local/data3/cmsdataops/srv/wmagent/v0.9.59/install/wmagent/PhEDExInjector/.wmcore_cache/.wmcore_cache_42935/requests/cmsweb.cern.ch/-6980513217279174487_POST_inject does not exist and the service at https://cmsweb.cern.ch/phedex/datasvc/json/prod/inject is unavailable - it returned 400 because Bad Request

This happens around once per week, and it needs to be fixed manually, using the script here:

https://raw.github.com/dballesteros7/dev-scripts/master/CMS-2012-2013/src/PhEDEx/fixPhEDEx.py

Can we do stop this from happening or fix it automatically?

Thanks.

Andrew


Changed files:
src/python/WMComponent/DBSBuffer/Database/MySQL/DBSBufferFiles/SetPhEDExStatus.py
src/python/WMCore/Services/PhEDEx/PhEDEx.py
src/python/WMComponent/PhEDExInjector/PhEDExInjectorPoller.py
src/python/WMComponent/DBSBuffer/Database/Oracle/DBSBufferFiles/SetPhEDExStatus.py

diff --git a/src/python/WMComponent/DBSBuffer/Database/MySQL/DBSBufferFiles/SetPhEDExStatus.py b/src/python/WMComponent/DBS3Buffer/MySQL/DBSBufferFiles/SetPhEDExStatus.py
similarity index 100%
rename from src/python/WMComponent/DBSBuffer/Database/MySQL/DBSBufferFiles/SetPhEDExStatus.py
rename to src/python/WMComponent/DBS3Buffer/MySQL/DBSBufferFiles/SetPhEDExStatus.py
diff --git a/src/python/WMComponent/DBSBuffer/Database/Oracle/DBSBufferFiles/SetPhEDExStatus.py b/src/python/WMComponent/DBS3Buffer/Oracle/DBSBufferFiles/SetPhEDExStatus.py
similarity index 67%
rename from src/python/WMComponent/DBSBuffer/Database/Oracle/DBSBufferFiles/SetPhEDExStatus.py
rename to src/python/WMComponent/DBS3Buffer/Oracle/DBSBufferFiles/SetPhEDExStatus.py
index b95c90c13..65de25018 100644
--- a/src/python/WMComponent/DBSBuffer/Database/Oracle/DBSBufferFiles/SetPhEDExStatus.py
+++ b/src/python/WMComponent/DBS3Buffer/Oracle/DBSBufferFiles/SetPhEDExStatus.py
@@ -5,7 +5,8 @@ _SetPhEDExStatus_
 Oracle implementation of DBSBufferFiles.SetPhEDExStatus
 """
 
-from WMComponent.DBSBuffer.Database.MySQL.DBSBufferFiles.SetPhEDExStatus import SetPhEDExStatus as MySQLSetPhEDExStatus
+from WMComponent.DBS3Buffer.MySQL.DBSBufferFiles.SetPhEDExStatus \\
+      import SetPhEDExStatus as MySQLSetPhEDExStatus
 
 class SetPhEDExStatus(MySQLSetPhEDExStatus):
     """
diff --git a/src/python/WMComponent/PhEDExInjector/PhEDExInjectorPoller.py b/src/python/WMComponent/PhEDExInjector/PhEDExInjectorPoller.py
index 8ef719cb6..1bd4a0c32 100644
--- a/src/python/WMComponent/PhEDExInjector/PhEDExInjectorPoller.py
+++ b/src/python/WMComponent/PhEDExInjector/PhEDExInjectorPoller.py
@@ -8,6 +8,7 @@ Poll the DBSBuffer database and inject files as they are created.
 import threading
 import logging
 import traceback
+from collections import defaultdict
 from httplib import HTTPException
 
 from WMCore.WorkerThreads.BaseWorkerThread import BaseWorkerThread
@@ -63,6 +64,7 @@ class PhEDExInjectorPoller(BaseWorkerThread):
         #    self.sendAlert will be then be available
         self.initAlerts(compName = "PhEDExInjector")
 
+        self.filesToRecover = None
 
     def setup(self, parameters):
         """
@@ -79,7 +81,7 @@ class PhEDExInjectorPoller(BaseWorkerThread):
         self.getUninjected = daofactory(classname = "GetUninjectedFiles")
         self.getMigrated = daofactory(classname = "GetMigratedBlocks")
 
-        daofactory = DAOFactory(package = "WMComponent.DBSBuffer.Database",
+        daofactory = DAOFactory(package = "WMComponent.DBS3Buffer",
                                 logger = self.logger,
                                 dbinterface = myThread.dbi)
         self.setStatus = daofactory(classname = "DBSBufferFiles.SetPhEDExStatus")
@@ -104,7 +106,7 @@ class PhEDExInjectorPoller(BaseWorkerThread):
         """
         _createInjectionSpec_
 
-        Trasform the data structure returned from the database into an XML
+        Transform the data structure returned from the database into an XML
         string for the PhEDEx Data Service.  The injectionData parameter must be
         a dictionary keyed by dataset path.  Each dataset path will map to a
         list of blocks, each block being a dict.  The block dicts will have
@@ -132,6 +134,32 @@ class PhEDExInjectorPoller(BaseWorkerThread):
 
         return injectionSpec.save()
 
+    def createRecoveryFileFormat(self, unInjectedData):
+        """
+        _createRecoveryFileFormat_
+
+        Transform the data structure returned from database in to the dict format
+        for the PhEDEx Data Service.  The injectionData parameter must be
+        a dictionary keyed by dataset path.  
+        
+        unInjectedData format
+        {"dataset1":
+          {"block1": {"is-open": "y", "files":
+            [{"lfn": "lfn1", "size": 10, "checksum": {"cksum": "1234"}},
+             {"lfn": "lfn2", "size": 20, "checksum": {"cksum": "4321"}}]}}}
+        
+        returns
+        {"block1": set(["lfn1", "lfn2"])}
+        """
+        sortedBlocks = defaultdict(set)
+        for datasetPath in unInjectedData:
+            
+            for fileBlockName, fileBlock in unInjectedData[datasetPath].iteritems():
+                for fileDict in fileBlock["files"]:
+                    sortedBlocks[fileBlockName].add(fileDict["lfn"])
+                    
+        return sortedBlocks
+    
     def injectFiles(self):
         """
         _injectFiles_
@@ -182,6 +210,15 @@ class PhEDExInjectorPoller(BaseWorkerThread):
             xmlData = self.createInjectionSpec(uninjectedFiles[siteName])
             try:
                 injectRes = self.phedex.injectBlocks(location, xmlData)
+            except HTTPException, ex:
+                # If we get an HTTPException of certain types, raise it as an error
+                if ex.status == 400:
+                    # assume it is duplicate injection error. but if that is not the case
+                    # needs to be investigated
+                    self.filesToRecover = self.createRecoveryFileFormat(uninjectedFiles[siteName])
+                
+                msg = "PhEDEx injection failed with %s error: %s" % (ex.status, ex.result)
+                raise PhEDExInjectorPassableError(msg)
             except Exception, ex:
                 # If we get an error here, assume that it's temporary (it usually is)
                 # log it, and ignore it in the algorithm() loop
@@ -189,6 +226,7 @@ class PhEDExInjectorPoller(BaseWorkerThread):
                 msg += str(ex)
                 logging.error(msg)
                 logging.debug("Traceback: %s" % str(traceback.format_exc()))
+                
                 raise PhEDExInjectorPassableError(msg)
             logging.info("Injection result: %s" % injectRes)
 
@@ -290,6 +328,27 @@ class PhEDExInjectorPoller(BaseWorkerThread):
             myThread.transaction.commit()
         return
 
+    def recoverInjectedFiles(self):
+        """
+        When PhEDEx inject call timed out, run this function.
+        Since there are 3 min reponse time out in cmsweb, some times 
+        PhEDEx injection call times out even though the call succeeded
+        In that case run the recovery mode
+        1. first check whether files which injection status = 0 are in the PhEDEx.
+        2. if those file exist set the in_phedex status to 1
+        3. set self.filesToRecover = None
+        """
+        myThread = threading.currentThread()
+        
+        injectedFiles = self.phedex.getInjectedFiles(self.filesToRecover)
+        
+        myThread.transaction.begin()
+        self.setStatus.execute(injectedFiles, 1)
+        myThread.transaction.commit()
+        # when files are recovered set the self.file 
+        self.filesToRecover = None
+        return injectedFiles
+        
     def algorithm(self, parameters):
         """
         _algorithm_
@@ -299,6 +358,13 @@ class PhEDExInjectorPoller(BaseWorkerThread):
         """
         myThread = threading.currentThread()
         try:
+            if self.filesToRecover != None:
+                logging.info(""" Running PhEDExInjector Recovery: 
+                                 previous injection call failed, 
+                                 check if files were injected to PhEDEx anyway""")
+                recoveredFiles = self.recoverInjectedFiles()
+                logging.info("%s files already injected: changed status in dbsbuffer db" % len(recoveredFiles))
+                        
             self.injectFiles()
             self.closeBlocks()
         except PhEDExInjectorPassableError, ex:
diff --git a/src/python/WMCore/Services/PhEDEx/PhEDEx.py b/src/python/WMCore/Services/PhEDEx/PhEDEx.py
index 87c082193..db47f1557 100644
--- a/src/python/WMCore/Services/PhEDEx/PhEDEx.py
+++ b/src/python/WMCore/Services/PhEDEx/PhEDEx.py
@@ -417,3 +417,24 @@ class PhEDEx(Service):
         # This is a dummy function to use in unittests to make sure the right class is
         # instantiated
         pass
+    
+    def getInjectedFiles(self, blockFileDict):
+        """
+        take dict of the input
+        {'block_name1': [file_lfn1, file_lfn2, ....],
+         'block_name2': [file_lfn1, file_lfn2, ....],           
+        }
+        and returns
+        list of file injected
+        """
+        injectedFiles = []
+        for block in blockFileDict:
+            result = self._getResult('data', args = {'block' : block}, verb = 'GET')
+            for dbs in result['phedex']['dbs']:
+                for dataset in dbs['dataset']:
+                    blockChunk = dataset['block']
+                    for blockInfo in blockChunk:
+                        for fileInfo in blockInfo['file']:
+                            if fileInfo['lfn'] in blockFileDict[block]:
+                                injectedFiles.append(fileInfo['lfn'])
+        return injectedFiles
\\ No newline at end of file