Merge pull request #10263 from amaltaro/bundle-145patch1

Bundle of patches for WMAgent 1.4.5.patch1 release

bin/adhoc-scripts/checkDsetFileCount.py

@@ -18,7 +18,7 @@
 RUCIO_ACCT = "wma_prod"
 RUCIO_HOST = "http://cms-rucio.cern.ch"
 RUCIO_AUTH = "https://cms-rucio-auth.cern.ch"
-DBS_URL = "https://cmsweb.cern.ch/dbs/prod/global/DBSReader/"
+DBS_URL = "https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader/"
 
 
 def loggerSetup(logLevel=logging.INFO):

bin/adhoc-scripts/workflowCompletion.py

@@ -150,7 +150,7 @@ def main():
         parser.error("You must provide either a workflow name or an input file name.")
         sys.exit(3)
 
-    cmswebUrl = "https://" + args.cms if args.cms else "https://cmsweb.cern.ch"
+    cmswebUrl = "https://" + args.cms if args.cms else "https://cmsweb-prod.cern.ch"
     reqmgrUrl = "https://" + args.reqmgr if args.reqmgr else "https://cmsweb.cern.ch"
 
     for reqName in listRequests:

bin/createStoreResults.py

@@ -48,7 +48,7 @@
     "AcquisitionEra": "UPDATEME",
     "ProcessingVersion": 1,  # will be updated too
     "Campaign": "StoreResults",
-    "DbsUrl": "https://cmsweb.cern.ch/dbs/prod/phys03/DBSReader",
+    "DbsUrl": "https://cmsweb-prod.cern.ch/dbs/prod/phys03/DBSReader",
     "GlobalTag": "crab3_tag",
     "Memory": 2000,
     "RequestPriority": 999999,
@@ -86,7 +86,7 @@ def migrateDataset(dset, dbsInst):
     Migrate dataset from the user instance to the DBS prod one.
     It returns the origin site name, which should be used for assignment
     """
-    dbsInst = "https://cmsweb.cern.ch/dbs/prod/%s/DBSReader" % dbsInst
+    dbsInst = "https://cmsweb-prod.cern.ch/dbs/prod/%s/DBSReader" % dbsInst
     migrateArgs = {'migration_url': dbsInst, 'migration_input': dset}
     dbsApi.submitMigration(migrateArgs)
     print("Migrating dataset %s from %s to prod/global" % (dset, dbsInst))
@@ -104,7 +104,7 @@ def buildRequest(userDict):
 
     newSchema = copy(DEFAULT_DICT)
     newSchema.update(userDict)
-    newSchema['DbsUrl'] = "https://cmsweb.cern.ch/dbs/prod/%s/DBSReader" % newSchema['DbsUrl']
+    newSchema['DbsUrl'] = "https://cmsweb-prod.cern.ch/dbs/prod/%s/DBSReader" % newSchema['DbsUrl']
     # Remove spaces from the Physics Group value
     newSchema['PhysicsGroup'] = newSchema['PhysicsGroup'].replace(" ", "")
     # Set PrepID according to the date and time

bin/fix-dbs-parentage

@@ -106,7 +106,7 @@ if __name__ == '__main__':
         end = time.mktime(datetime.datetime.strptime(args.end, "%d/%m/%Y").timetuple())
 
     if args.prod:
-        dbsURL = 'https://cmsweb.cern.ch/dbs/prod/global/DBSWriter'
+        dbsURL = 'https://cmsweb-prod.cern.ch/dbs/prod/global/DBSWriter'
         logger.info("Using production: %s", dbsURL)
     else:
         dbsURL = 'https://cmsweb-testbed.cern.ch/dbs/int/global/DBSWriter'

deploy/WMAgent.production

@@ -15,7 +15,7 @@ WMSTATS_URL=https://cmsweb.cern.ch/couchdb/wmstats
 REQMGR_URL=https://cmsweb.cern.ch/reqmgr/rest
 ACDC_URL=https://cmsweb.cern.ch/couchdb/acdcserver
 WORKLOAD_SUMMARY_URL=https://cmsweb.cern.ch/couchdb/workloadsummary
-DBS3_URL=https://cmsweb.cern.ch/dbs/prod/global/DBSWriter
+DBS3_URL=https://cmsweb-prod.cern.ch/dbs/prod/global/DBSWriter
 PHEDEX_URL=https://cmsweb.cern.ch/phedex/datasvc/json/prod/
 DQM_URL=https://cmsweb.cern.ch/dqm/offline
 DASHBOARD_URL=http://dashb-ssb.cern.ch/dashboard

deploy/deploy-wmagent.sh

@@ -24,8 +24,8 @@
 ### Usage:               -n <agent_number> Agent number to be set when more than 1 agent connected to the same team (defaults to 0)
 ### Usage:
 ### Usage: deploy-wmagent.sh -w <wma_version> -d <deployment_tag> -t <team_name> [-s <scram_arch>] [-r <repository>] [-n <agent_number>]
-### Usage: Example: sh deploy-wmagent.sh -w 1.4.3.patch2 -d HG2012f -t production -n 30
-### Usage: Example: sh deploy-wmagent.sh -w 1.4.3.patch2 -d HG2012f -t testbed-vocms001 -p "9963 9959" -r comp=comp.amaltaro
+### Usage: Example: sh deploy-wmagent.sh -w 1.4.5.patch1 -d HG2102e -t production -n 30
+### Usage: Example: sh deploy-wmagent.sh -w 1.4.5.patch2 -d HG2102e -t testbed-vocms001 -p "9963 9959" -r comp=comp.amaltaro
 ### Usage:
 
 IAM=`whoami`
@@ -342,8 +342,8 @@ if [[ "$TEAMNAME" == relval ]]; then
   sed -i "s+config.RucioInjector.metaDIDProject = 'Production'+config.RucioInjector.metaDIDProject = 'RelVal'+" $MANAGE_DIR/config.py
 elif [[ "$TEAMNAME" == *testbed* ]] || [[ "$TEAMNAME" == *dev* ]]; then
   GLOBAL_DBS_URL=https://cmsweb-testbed.cern.ch/dbs/int/global/DBSReader
-  sed -i "s+DBSInterface.globalDBSUrl = 'https://cmsweb.cern.ch/dbs/prod/global/DBSReader'+DBSInterface.globalDBSUrl = '$GLOBAL_DBS_URL'+" $MANAGE_DIR/config.py
-  sed -i "s+DBSInterface.DBSUrl = 'https://cmsweb.cern.ch/dbs/prod/global/DBSReader'+DBSInterface.DBSUrl = '$GLOBAL_DBS_URL'+" $MANAGE_DIR/config.py
+  sed -i "s+DBSInterface.globalDBSUrl = 'https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader'+DBSInterface.globalDBSUrl = '$GLOBAL_DBS_URL'+" $MANAGE_DIR/config.py
+  sed -i "s+DBSInterface.DBSUrl = 'https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader'+DBSInterface.DBSUrl = '$GLOBAL_DBS_URL'+" $MANAGE_DIR/config.py
   sed -i "s+config.RucioInjector.metaDIDProject = 'Production'+config.RucioInjector.metaDIDProject = 'Test'+" $MANAGE_DIR/config.py
 fi
 

doc/createSpecs/DQMHarvest_createSpec.json

@@ -75,7 +75,7 @@
         "type": "<type 'int'>"
     }, 
     "DbsUrl": {
-        "default": "https://cmsweb.cern.ch/dbs/prod/global/DBSReader", 
+        "default": "https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader",
         "optional": true, 
         "type": "<type 'str'>"
     }, 

doc/createSpecs/ReReco_createSpec.json

@@ -75,7 +75,7 @@
         "type": "<type 'int'>"
     }, 
     "DbsUrl": {
-        "default": "https://cmsweb.cern.ch/dbs/prod/global/DBSReader", 
+        "default": "https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader",
         "optional": true, 
         "type": "<type 'str'>"
     }, 

doc/createSpecs/StepChain_createSpec.json

@@ -70,7 +70,7 @@
         "type": "<type 'int'>"
     }, 
     "DbsUrl": {
-        "default": "https://cmsweb.cern.ch/dbs/prod/global/DBSReader", 
+        "default": "https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader",
         "optional": true, 
         "type": "<type 'str'>"
     }, 

doc/createSpecs/StoreResults_createSpec.json

@@ -80,7 +80,7 @@
         "type": "<type 'str'>"
     }, 
     "DbsUrl": {
-        "default": "https://cmsweb.cern.ch/dbs/prod/global/DBSReader", 
+        "default": "https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader",
         "optional": true, 
         "type": "<type 'str'>"
     }, 

doc/createSpecs/TaskChain_createSpec.json

@@ -65,7 +65,7 @@
         "type": "<type 'int'>"
     }, 
     "DbsUrl": {
-        "default": "https://cmsweb.cern.ch/dbs/prod/global/DBSReader", 
+        "default": "https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader",
         "optional": true, 
         "type": "<type 'str'>"
     }, 

etc/WMAgentConfig.py

@@ -47,7 +47,7 @@
 
 # DBS Information.
 localDBSVersion = "DBS_2_0_8"
-globalDBSUrl = "https://cmsweb.cern.ch/dbs/prod/global/DBSReader"
+globalDBSUrl = "https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader"
 globalDBSVersion = "DBS_2_0_8"
 
 # List of SE for T1 _Disk endpoints (TODO clean this up at some point)
@@ -145,7 +145,7 @@
 config.DBS3Upload.logLevel = globalLogLevel
 config.DBS3Upload.workerThreads = 1
 config.DBS3Upload.pollInterval = 100
-# "https://cmsweb.cern.ch/dbs/prod/global/DBSWriter" - production one
+# "https://cmsweb-prod.cern.ch/dbs/prod/global/DBSWriter" - production one
 config.DBS3Upload.dbsUrl = "OVERWRITE_BY_SECRETS"
 config.DBS3Upload.primaryDatasetType = "mc"
 config.DBS3Upload.dumpBlock = False  # to dump block meta-data into a json file

etc/dbsVerify.py

@@ -59,7 +59,7 @@ def connectToDB():
     blocks[row[0]].append(row[1])
 
 args = {}
-args["url"] = "https://cmsweb.cern.ch/dbs/prod/global/DBSReader"
+args["url"] = "https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader"
 args["version"] = "DBS_2_0_9"
 args["mode"] = "GET"
 dbsApi = DbsApi(args)

etc/harvestingInjector.py

@@ -94,7 +94,7 @@ def injectFilesFromDBS(inputFileset, datasetPath, runsWhiteList=[]):
     """
     print("injecting files from %s into %s, please wait..." % (datasetPath, inputFileset.name))
     args = {}
-    args["url"] = "https://cmsweb.cern.ch/dbs/prod/global/DBSReader"
+    args["url"] = "https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader"
     args["version"] = "DBS_2_1_1"
     args["mode"] = "GET"
     dbsApi = DbsApi(args)

etc/injectReRecoWorkflow.py

@@ -65,7 +65,7 @@ def injectFilesFromDBS(inputFileset, datasetPath):
     """
     print("injecting files from %s into %s, please wait..." % (datasetPath, inputFileset.name))
     args={}
-    args["url"] = "https://cmsweb.cern.ch/dbs/prod/global/DBSReader"
+    args["url"] = "https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader"
     args["version"] = "DBS_2_0_9"
     args["mode"] = "GET"
     dbsApi = DbsApi(args)

etc/injectStoreResults.py

@@ -77,7 +77,7 @@ def injectFilesFromDBS(inputFileset, datasetPath):
     """
     print("injecting files from %s into %s, please wait..." % (datasetPath, inputFileset.name))
     args={}
-    args["url"] = "https://cmsweb.cern.ch/dbs/prod/global/DBSReader"
+    args["url"] = "https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader"
     args["mode"] = "GET"
     dbsApi = DbsApi(args)
     dbsResults = dbsApi.listFileArray(path = datasetPath, retriveList = ["retrive_lumi", "retrive_run"])

etc/submit.sh

@@ -96,19 +96,23 @@ echo -e "======== WMAgent validate arguments finished at $(TZ=GMT date) ========
 echo "======== WMAgent CMS environment load starting at $(TZ=GMT date) ========"
 if [ -f "$VO_CMS_SW_DIR"/cmsset_default.sh ]
 then  #   LCG style --
+    echo "WN with a LCG style environment, thus using VO_CMS_SW_DIR=$VO_CMS_SW_DIR"
     . $VO_CMS_SW_DIR/cmsset_default.sh
 elif [ -f "$OSG_APP"/cmssoft/cms/cmsset_default.sh ]
 then  #   OSG style --
+    echo "WN with an OSG style environment, thus using OSG_APP=$OSG_APP"
     . $OSG_APP/cmssoft/cms/cmsset_default.sh CMSSW_3_3_2
 elif [ -f "$CVMFS"/cms.cern.ch/cmsset_default.sh ]
 then
+    echo "WN with CVMFS environment, thus using CVMFS=$CVMFS"
     . $CVMFS/cms.cern.ch/cmsset_default.sh
 elif [ -f /cvmfs/cms.cern.ch/cmsset_default.sh ]
 then  # ok, lets call it CVMFS then
     export CVMFS=/cvmfs/cms.cern.ch
+    echo "WN missing VO_CMS_SW_DIR/OSG_APP/CVMFS environment variable, forcing it to CVMFS=$CVMFS"
     . $CVMFS/cmsset_default.sh
 else
-    echo "Error during job bootstrap: VO_CMS_SW_DIR, OSG_APP, CVMFS  or /cvmfs were not found." >&2
+    echo "Error during job bootstrap: VO_CMS_SW_DIR, OSG_APP, CVMFS or /cvmfs were not found." >&2
     echo "  Because of this, we can't load CMSSW. Not good." >&2
     exit 11003
 fi
@@ -137,8 +141,8 @@ elif [ -d "$CVMFS"/COMP/"$WMA_SCRAM_ARCH"/external/python ]
 then
     prefix="$CVMFS"/COMP/"$WMA_SCRAM_ARCH"/external/python
 else
-    echo "Error during job bootstrap: job environment does not contain the init.sh script." >&2
-    echo "  Because of this, we can't load CMSSW. Not good." >&2
+    echo "Failed to find a COMP python installation in the worker node setup." >&2
+    echo "  Without a known python, there is nothing else we can do with this job. Quiting!" >&2
     exit 11004
 fi
 

src/python/Utils/Utilities.py

@@ -224,6 +224,34 @@ def decodeBytesToUnicode(value, errors="strict"):
     return value
 
 
+def encodeUnicodeToBytes(value, errors="strict"):
+    """
+    Accepts an input "value" of generic type.
+
+    If "value" is a string of type sequence of unicode (i.e. in py2 `unicode` or
+    `future.types.newstr.newstr`, in py3 `str`), then it is converted to
+    a sequence of bytes.
+
+    This function is useful for encoding output data when using the
+    "unicode sandwich" approach, which involves converting unicode (i.e. strings
+    of type sequence of unicode codepoints) to bytes (i.e. strings of type
+    sequence of bytes, in py2 `str` or `future.types.newbytes.newbytes`,
+    in py3 `bytes`) as late as possible when passing a string to a third-party
+    function that only accepts bytes as input (pycurl's curl.setop is an
+    example).
+    py2:
+    - "errors" can be: "strict", "ignore", "replace", "xmlcharrefreplace"
+    - ref: https://docs.python.org/2/howto/unicode.html#the-unicode-type
+    py3:
+    - "errors" can be: "strict", "ignore", "replace", "backslashreplace", 
+      "xmlcharrefreplace", "namereplace"
+    - ref: https://docs.python.org/3/howto/unicode.html#the-string-type
+    """
+    if isinstance(value, str):
+        return value.encode("utf-8", errors)
+    return value
+
+
 # TODO: remove this function once we have completely migrated to Rucio
 def usingRucio():
     """

src/python/WMComponent/DBS3Buffer/DBSUploadPoller.py

@@ -48,38 +48,6 @@
 from WMCore.WorkerThreads.BaseWorkerThread import BaseWorkerThread
 
 
-def createConfigForJSON(config):
-    """
-    Turn a config object into a dictionary of dictionaries
-
-    """
-
-    final = {}
-    for sectionName in config.listSections_():
-        section = getattr(config, sectionName)
-        if hasattr(section, 'dictionary_'):
-            # Create a dictionary key for it
-            final[sectionName] = createDictionaryFromConfig(section)
-
-    return final
-
-
-def createDictionaryFromConfig(configSection):
-    """
-    Recursively create dictionaries from config
-
-    """
-
-    final = configSection.dictionary_()
-
-    for key in final.keys():
-        if hasattr(final[key], 'dictionary_'):
-            # Then we can turn it into a dictionary
-            final[key] = createDictionaryFromConfig(final[key])
-
-    return final
-
-
 def uploadWorker(workInput, results, dbsUrl):
     """
     _uploadWorker_
@@ -141,11 +109,35 @@ def uploadWorker(workInput, results, dbsUrl):
     return
 
 
+def isPassiveError(exceptionObj):
+    """
+    This function will parse the exception object and report whether
+    the error message corresponds to a soft or hard error (hard errors
+    are supposed to let the component crash).
+    :param exceptionObj: any exception object
+    :return: True if it's a soft error, False otherwise
+    """
+    passException = True
+    passiveErrorMsg = ['Service Unavailable', 'Service Temporarily Unavailable',
+                       'Proxy Error', 'Error reading from remote server',
+                       'Connection refused', 'timed out', 'Could not resolve',
+                       'OpenSSL SSL_connect: SSL_ERROR_SYSCALL']
+
+    excReason = getattr(exceptionObj, 'reason', '')
+    for passiveMsg in passiveErrorMsg:
+        if passiveMsg in excReason:
+            break
+        elif passiveMsg in str(exceptionObj):
+            break
+    else:
+        passException = False
+    return passException
+
+
 class DBSUploadException(WMException):
     """
     Holds the exception info for
     all the things that will go wrong
-
     """
 
 
@@ -344,19 +336,16 @@ def updateDatasetParentageCache(self):
         try:
             self.datasetParentageCache = self.wmstatsServerSvc.getChildParentDatasetMap()
         except Exception as ex:
-            reason = getattr(ex, 'reason', '')
-            msg = 'Failed to fetch parentage map from WMStats, skipping this cycle. '
-            msg += 'Exception: %s. Reason: %s. ' % (type(ex).__name__, reason)
-            if 'Service Unavailable' in reason or 'Proxy Error' in reason or \
-                            'Error reading from remote server' in reason:
-                pass
-            elif 'Connection refused' in str(ex) or 'timed out' in str(ex) or 'Could not resolve' in str(ex):
-                msg += 'Error: %s' % str(ex)
+            excReason = getattr(ex, 'reason', '')
+            errorMsg = 'Failed to fetch parentage map from WMStats, skipping this cycle. '
+            errorMsg += 'Exception: {}. Reason: {}. Error: {}. '.format(type(ex).__name__,
+                                                                        excReason, str(ex))
+            if isPassiveError(ex):
+                logging.warning(errorMsg)
             else:
-                msg = "Unknown failure while fetching parentage map from WMStats. Error: %s" % str(ex)
-                raise DBSUploadException(msg)
-            logging.warning(msg)
-            myThread.logdbClient.post("DBS3Upload_parentMap", msg, "warning")
+                errorMsg += 'Hit a terminal exception in DBSUploadPoller.'
+                raise DBSUploadException(errorMsg)
+            myThread.logdbClient.post("DBS3Upload_parentMap", errorMsg, "warning")
             success = False
         else:
             myThread.logdbClient.delete("DBS3Upload_parentMap", "warning", this_thread=True)
@@ -376,7 +365,7 @@ def loadBlocks(self):
         # Load them if we don't have them
         blocksToLoad = []
         for block in openBlocks:
-            if block['blockname'] not in self.blockCache.keys():
+            if block['blockname'] not in self.blockCache:
                 blocksToLoad.append(block['blockname'])
 
         # Now load the blocks

src/python/WMComponent/RucioInjector/RucioInjectorPoller.py

@@ -287,6 +287,10 @@ def insertBlockRules(self):
             if not self._isBlockTierAllowed(item['blockname']):
                 logging.debug("Component configured to skip block rule for: %s", item['blockname'])
                 continue
+            # first, check if the block has already been created in Rucio
+            if not self.rucio.didExist(item['blockname']):
+                logging.warning("Block: %s not yet in Rucio. Retrying later..", item['blockname'])
+                continue
             kwargs = dict(activity="Production Output", account=self.rucioAcct,
                           grouping="DATASET", comment="WMAgent automatic container rule",
                           ignore_availability=True, meta=self.metaData)
@@ -519,6 +523,10 @@ def insertContainerRules(self):
                              container, rseName)
                 subscriptionsMade.append(subInfo['id'])
                 continue
+            # then check if the container has already been created in Rucio
+            if not self.rucio.didExist(container):
+                logging.warning("Container: %s not yet in Rucio. Retrying later..", container)
+                continue
 
             ruleKwargs = dict(ask_approval=False,
                               activity=self._activityMap(rseName),