Add DBS3_READER_URL and WorkflowUpdater.dbsUrl

bin/wmagent-mod-config

@@ -288,13 +288,15 @@ def modifyConfiguration(config, **args):
     if hasattr(config, "ArchiveDataReporter"):
         config.ArchiveDataReporter.WMArchiveURL = args["wmarchive_url"]
 
-        # custom WMArchiveReporter
+    # custom WMArchiveReporter
     if hasattr(config, "WorkflowUpdater"):
+        config.WorkflowUpdater.dbsUrl = args.get("dbs3_reader_url", None)
         config.WorkflowUpdater.rucioUrl = args["rucio_host"]
         config.WorkflowUpdater.rucioAuthUrl = args["rucio_auth"]
         if args.get("mspileup_url", None):
             config.WorkflowUpdater.msPileupUrl = args["mspileup_url"]
 
+
     return config
 
 
@@ -319,7 +321,7 @@ def main(argv=None):
                                          "sb_hostname=", "sb_port=", "sb_basepath=", "ufc_hostname=", "ufc_port=",
                                          "ufc_service_url=", "ufc_cachedir=", "client_mapping=",
                                          "workload_summary_url=", "coredb_url=", "wmstats_url=", "ops_proxy=",
-                                         "reqmgr2_url=", "acdc_url=", "amq_auth_file=", "dbs3_url=",
+                                         "reqmgr2_url=", "acdc_url=", "amq_auth_file=", "dbs3_url=", "dbs3_reader_url=",
                                          "dqm_url=", "grafana_token=", "requestcouch_url=", "central_logdb_url=",
                                          "wmarchive_url=", "amq_credentials=",
                                          "rucio_account=", "rucio_host=", "rucio_auth=", "mspileup_url="])
@@ -343,7 +345,7 @@ def main(argv=None):
                           '--ufc_port', '--ufc_service_url', '--ufc_cachedir',
                           '--client_mapping', '--workload_summary_url',
                           '--wmstats_url', '--ops_proxy', '--reqmgr2_url', '--acdc_url',
-                          '--amq_auth_file', '--dbs3_url', '--dqm_url',
+                          '--amq_auth_file', '--dbs3_url', '--dbs3_reader_url', '--dqm_url',
                           '--grafana_token', '--requestcouch_url', '--central_logdb_url',
                           '--wmarchive_url', '--amq_credentials',
                           '--rucio_account', '--rucio_host', '--rucio_auth', '--mspileup_url'):

deploy/WMAgent.production

@@ -15,6 +15,7 @@ WMSTATS_URL=https://cmsweb.cern.ch/couchdb/wmstats
 ACDC_URL=https://cmsweb.cern.ch/couchdb/acdcserver
 WORKLOAD_SUMMARY_URL=https://cmsweb.cern.ch/couchdb/workloadsummary
 DBS3_URL=https://cmsweb-prod.cern.ch/dbs/prod/global/DBSWriter
+DBS3_READER_URL=https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader
 DQM_URL=https://cmsweb.cern.ch/dqm/offline
 REQUESTCOUCH_URL=https://cmsweb.cern.ch/couchdb/reqmgr_workload_cache
 REQMGR2_URL=https://cmsweb.cern.ch/reqmgr2

deploy/WMAgent.testbed

@@ -15,6 +15,7 @@ WMSTATS_URL=https://cmsweb-testbed.cern.ch/couchdb/wmstats
 ACDC_URL=https://cmsweb-testbed.cern.ch/couchdb/acdcserver
 WORKLOAD_SUMMARY_URL=https://cmsweb-testbed.cern.ch/couchdb/workloadsummary
 DBS3_URL=https://cmsweb-testbed.cern.ch/dbs/int/global/DBSWriter
+DBS3_READER_URL=https://cmsweb-testbed.cern.ch/dbs/int/global/DBSReader
 DQM_URL=https://cmsweb-testbed.cern.ch/dqm/dev
 REQUESTCOUCH_URL=https://cmsweb-testbed.cern.ch/couchdb/reqmgr_workload_cache
 REQMGR2_URL=https://cmsweb-testbed.cern.ch/reqmgr2

etc/WMAgentConfig.py

@@ -391,6 +391,7 @@
 config.WorkflowUpdater.componentDir = config.General.workDir + "/WorkflowUpdater"
 config.WorkflowUpdater.logLevel = globalLogLevel
 config.WorkflowUpdater.pollInterval = 8 * 60 * 60  # every 8 hours
+config.WorkflowUpdater.dbsUrl = "OVER_WRITE_BY_SECRETS"
 config.WorkflowUpdater.rucioAccount = "wmcore_pileup"
 config.WorkflowUpdater.rucioUrl = "OVER_WRITE_BY_SECRETS"
 config.WorkflowUpdater.rucioAuthUrl = "OVER_WRITE_BY_SECRETS"

src/python/WMComponent/WorkflowUpdater/WorkflowUpdaterPoller.py

@@ -118,7 +118,7 @@ def checkChanges(jdict, msPUBlockLoc):
     return False
 
 
-def updateBlockInfo(jdoc, msPUBlockLoc, logger):
+def updateBlockInfo(jdoc, msPUBlockLoc, dbsUrl, logger):
     """
     Update block information within sandbox json pileup conf file using MSPileup block location info.
     The main logic is the following:
@@ -143,6 +143,7 @@ def updateBlockInfo(jdoc, msPUBlockLoc, logger):
 
     :param jdoc: JSON sandbox dictionary
     :param msPUBlockLOck: dict of block with rses from MSPileup service, i.e. {'block': [rses], ... }
+    :param dbsUrl: dbs Url
     :param logger: logger object
     :return: newly constructed dict
     """
@@ -178,7 +179,7 @@ def updateBlockInfo(jdoc, msPUBlockLoc, logger):
     if len(blocksToUpdate) > 0:
         logger.info("Adding %s blocks from MSPileup which are not present in pileupconf.json",
                     len(blocksToUpdate))
-        binfo = getBlockInfo4PU(blocksToUpdate, ckey(), cert())
+        binfo = getBlockInfo4PU(blocksToUpdate, dbsUrl, ckey(), cert())
         for puType in returnDict.keys():
             for blockName in blocksToUpdate:
                 # update block record in-place
@@ -250,6 +251,10 @@ def __init__(self, config):
 
         # parse mandatory attributes from the configuration
         self.config = config
+        dbsUrl = "https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader"
+        self.dbsUrl = getattr(config.WorkflowUpdater, "dbsUrl", dbsUrl)
+        if not self.dbsUrl:
+            self.dbsUrl = dbsUrl
         self.rucioAcct = getattr(config.WorkflowUpdater, "rucioAccount")
         self.rucioUrl = getattr(config.WorkflowUpdater, "rucioUrl")
         self.rucioAuthUrl = getattr(config.WorkflowUpdater, "rucioAuthUrl")
@@ -385,13 +390,17 @@ def adjustJSONSpec(self, puWflows, msPileupList, dest=None):
                         # {block1: [rses], block2: [rses], ...}
                         if checkChanges(jsonBlockLoc, msPUBlockLoc):
                             self.logger.info("Found differences between JSON and MSPileup content.")
-                            puNewJsonContent = updateBlockInfo(puJsonContent, msPUBlockLoc, self.logger)
+                            puNewJsonContent = updateBlockInfo(puJsonContent, msPUBlockLoc, self.dbsUrl, self.logger)
 
                             # we should update a tarball only once for each pileup name,
                             # therefore we add new entry to jdict with our pilupe conf file
-                            jdict[puConfJson] = puNewJsonContent
-                            self.logger.info("Mark %s to be updated in tarball %s with a fresh pileup content",
-                                             puConfJson, tarFile)
+                            if puNewJsonContent:
+                                # we update json file if we get new pileup content
+                                jdict[puConfJson] = puNewJsonContent
+                                self.logger.info("Mark %s to be updated in tarball %s with a fresh pileup content",
+                                                 puConfJson, tarFile)
+                            else:
+                                self.logger.warning("updateBlockInfo did not return any results for %s, will skip update of pileup json content", pileupName)
                         else:
                             msg = "### There are no differences between JSON and MSPileup content "
                             msg += f"for pileup name {pileupName}. Not updating anything!"

src/python/WMCore/Services/DBS/DBSConcurrency.py

@@ -11,20 +11,23 @@
 from WMCore.Services.pycurl_manager import getdata as multi_getdata
 
 
-def getBlockInfo4PU(blockNames, ckey, cert):
+def getBlockInfo4PU(blockNames, dbsUrl, ckey, cert):
     """
     Fetch block information details, file list and number of events, from DBS
     server. Here we use concrete set of parameters for DBS to use in this case, i.e.
     we must look-up only valid files and get full details from the DBS API (in order
     to get number of events).
     :param blockNames: list of block names
+    :param dbsUrl: dbs URL
+    :param ckey: user keyfile
+    :param cert: user certificate
     :return: dictionary of {block: {"FileList": list of strings, "NumberOfEvents": integer}, ...}
     """
     urls = []
     for blk in blockNames:
         # need to encode block name properly
         block = urllib.parse.quote_plus(blk)
-        url = f"https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader/files?detail=true&validFileOnly=1&block_name={block}"
+        url = f"{dbsUrl}/files?detail=true&validFileOnly=1&block_name={block}"
         urls.append(url)
     # place concurrent calls to DBS, please note that multi_getdata is generator, therefore
     # it does not put DBS results into the memory until this generator is iterated