Resolve the premix file list once instead of in every job

.gitignore

@@ -12,3 +12,6 @@ __pycache__/
 
 # user-specific config (see docs/configuration/settings.md)
 /config/user_custom.yaml
+
+# crab drops one wherever it runs; the wrapper keeps it in $DSPROD_CRAB_HOME, this catches strays
+crab.log

docs/concepts/backends.md

@@ -172,6 +172,20 @@ the site they were assigned — CRAB cannot re-target a running task.
 
 A site you know is bad belongs in the static `blacklist` instead: that one is never lifted.
 
+!!! note "CRAB does not write to your AFS home"
+    CRAB rewrites its task cache `~/.crab3` on *every* command, status queries included. With
+    `$HOME` on AFS that makes a long production depend on an AFS token: when the token lapses,
+    every status query fails at once with
+    `PermissionError: [Errno 13] Permission denied: '/afs/.../.crab3.<pid>'` and law reports it as
+    a status-query failure for all jobs. The `crab` wrapper `env.sh` installs therefore points
+    `HOME` at `$DSPROD_CRAB_HOME` (default: a per-user directory under `$TMPDIR`), so nothing in a
+    production run needs AFS. law passes `--proxy` to submit, status and kill, so CRAB never needs
+    `~/.globus` from the real home either.
+
+    DSProd still renews Kerberos and the AFS token (`kinit -R` + `aklog`, hourly, from
+    `crab_poll_callback`) — but renewal can only extend a ticket that is still valid, so it is not
+    a substitute for keeping the production off AFS.
+
 !!! note "Why `env.sh` matters for CRAB"
     law runs `crab` inside a CMSSW sandbox of its own, and dumps that sandbox's environment with
     bare `python` — which modern CMSSW no longer ships, and for which the DSProd venv's `python`

docs/concepts/tasks.md

@@ -67,6 +67,25 @@ that is worth a batch backend (`--workflow htcondor|crab`).
     unreachable from the worker.
 
 
+### `PremixFileList`
+
+Resolves an era's premix pileup dataset to a plain file list on `fs_default`
+(`<output>/premix/<era>.txt`), once, and `RunProd` passes it to `cmsDriver` as
+`--pileup_input filelist:...`.
+
+Without it, `--pileup_input dbs:<dataset>` makes **every job** resolve that dataset — ~38 000 files
+— with its own DAS query. At a few thousand concurrent jobs the queries start returning nothing,
+cmsDriver then writes a config with no secondary input, and cmsRun dies on
+
+```
+NoSecondaryFiles: RootEmbeddedFileSequence no input files specified for secondary input source
+```
+
+*after* the job has already produced its GEN-SIM. A 5000-job production lost 10 % of its jobs that
+way. The list is identical to what a successful DAS query returns, so nothing about the physics
+changes; it is stored in the production area (not `<output>_test`) because it depends only on the
+era, and a batch node refuses to build one — that query belongs on the submitting machine.
+
 ### `RunProd`
 
 The core production task: a fused GEN→…→MiniAOD→NanoAOD chain for one `(era, point, seed)`, run

dsprod/crab.py

@@ -20,10 +20,14 @@
       # refill_fraction: 0.2     # min wave size / free slots, as a fraction of parallel_jobs
 """
 
+import fnmatch
+import json
 import math
 import os
 import re
 import subprocess
+import time
+import urllib.request
 import uuid
 from collections import Counter, OrderedDict
 
@@ -192,6 +196,85 @@ def _rewrite_crab_job_file(job_file):
             f.writelines(new_lines)
 
 
+#: CRIC's site table — the same source CRAB validates a whitelist against
+_CRIC_URL = "https://cms-cric.cern.ch/api/cms/site/query/?json"
+
+#: how long a cached site list is reused before CRIC is asked again
+_CRIC_CACHE_SECONDS = 24 * 3600
+
+
+def processing_sites(cache_path=None, url=_CRIC_URL, timeout=60):
+    """CMS site names that actually run jobs, newest-first from CRIC, cached on disk.
+
+    `/cvmfs/cms.cern.ch/SITECONF` cannot be used for this: it also lists storage endpoints such as
+    `T1_US_FNAL_Disk` and `T3_CH_CERNBOX`, and a whitelist naming one gets the task refused --
+    "A site name T1_US_FNAL_Disk that user specified is not in the list of known CMS Processing
+    Site Names". CRIC marks the difference: a site that runs jobs has `computeunits`.
+    """
+    if cache_path and os.path.exists(cache_path):
+        if time.time() - os.path.getmtime(cache_path) < _CRIC_CACHE_SECONDS:
+            try:
+                with open(cache_path) as f:
+                    return json.load(f)
+            except (OSError, ValueError):
+                pass
+    try:
+        with urllib.request.urlopen(url, timeout=timeout) as response:
+            payload = json.load(response)
+    except Exception as exc:
+        if cache_path and os.path.exists(cache_path):
+            with open(cache_path) as f:  # stale is better than nothing
+                return json.load(f)
+        raise RuntimeError(f"could not read the CMS site list from {url}: {exc}")
+    entries = payload.values() if isinstance(payload, dict) else payload
+    sites = sorted(
+        e["name"] for e in entries if e.get("name") and e.get("computeunits")
+    )
+    if cache_path:
+        try:
+            os.makedirs(os.path.dirname(cache_path), exist_ok=True)
+            with open(cache_path, "w") as f:
+                json.dump(sites, f)
+        except OSError:
+            pass
+    return sites
+
+
+def resolve_whitelist(whitelist, blacklist, sites):
+    """A `Site.whitelist` from which `blacklist` is actually absent.
+
+    CRAB gives the whitelist precedence: a site matched by both is *kept*, and it says so only in a
+    warning ("Since the whitelist has precedence, these sites are not considered in the blacklist").
+    With the default all-tier globs that silently defeats every exclusion -- the configured
+    `crab.blacklist` and the automatic site quarantine alike.
+
+    So a tier glob covering an excluded site is expanded, from `sites`, into the sites it actually
+    matches minus the excluded ones. Globs covering nothing excluded are left alone, which keeps the
+    pool wide and the expansion small: excluding one T2 lists the T2s and leaves `T1_*` and `T3_*`
+    as they are.
+    """
+    if not blacklist:
+        return list(whitelist)
+    out = []
+    for entry in whitelist:
+        hit = [b for b in blacklist if fnmatch.fnmatch(b, entry)]
+        if not hit:
+            out.append(entry)
+            continue
+        # an entry that is itself excluded simply disappears
+        out += [
+            site
+            for site in sites
+            if fnmatch.fnmatch(site, entry) and site not in blacklist
+        ]
+    if not out:
+        raise RuntimeError(
+            f"the blacklist {', '.join(blacklist)} excludes every site the whitelist "
+            f"{', '.join(whitelist)} allows"
+        )
+    return out
+
+
 _CrabProxyBase = law.cms.CrabWorkflow.workflow_proxy_cls
 
 
@@ -450,7 +533,9 @@ def crab_poll_callback(self, poll_data):
         if self._crab_kerberos_update is None:
 
             def renew_kerberos_ticket():
-                update_kerberos_ticket(verbose=0)
+                # verbose: a silent renewal leaves no way to tell, after a credential failure,
+                # whether it had been running at all
+                update_kerberos_ticket(verbose=1)
 
             krenew = float(getattr(self, "krenew", 1) or 0)
             self._crab_kerberos_update = (
@@ -592,7 +677,12 @@ def crab_job_config(self, config, job_nums, branches=None):
             )
             blacklist = list(blacklist) + quarantined
 
-        config.crab.Site.whitelist = [str(s) for s in whitelist or _CRAB_ALL_SITES]
+        sites = resolve_whitelist(
+            whitelist or _CRAB_ALL_SITES,
+            blacklist,
+            processing_sites(os.path.join(self.ana_data_path(), "cms_sites.json")),
+        )
+        config.crab.Site.whitelist = [str(s) for s in sites]
         if blacklist:
             config.crab.Site.blacklist = [str(s) for s in blacklist]
         # Keep CMS's global blacklist of known-broken sites in force unless explicitly waived:

dsprod/run_step.py

@@ -93,6 +93,7 @@ def build_cmsdriver(
     gridpack=None,
     fragment_rel=None,
     n_threads=1,
+    pileup_filelist=None,
 ):
     """Assemble the cmsDriver.py command line for one step."""
     out = fileout or f"{step}.root"
@@ -118,6 +119,10 @@ def build_cmsdriver(
     # derives numberOfStreams from it. A single-threaded cmsRun in a multi-core slot wastes
     # the extra cores and does not get any faster.
     cmd += f" --mc -n {n_evt} --nThreads {int(step_params.get('nThreads', n_threads))}"
+    pileup = step_params.get("pileup_input")
+    if pileup and pileup_filelist and str(pileup).startswith(("dbs:", "das:")):
+        # resolved once by PremixFileList instead of by a DAS query in every job
+        pileup = f"filelist:{pileup_filelist}"
     # empty means "none": an era can drop a modifier that `default_step` sets for the others
     if step_params.get("procModifiers"):
         cmd += f" --procModifiers {step_params['procModifiers']}"
@@ -127,8 +132,8 @@ def build_cmsdriver(
         cmd += f" --datamix {step_params['datamix']}"
     if "pileup" in step_params:
         cmd += f" --pileup {step_params['pileup']}"
-    if "pileup_input" in step_params:
-        cmd += f" --pileup_input \"{step_params['pileup_input']}\""
+    if pileup:
+        cmd += f' --pileup_input "{pileup}"' ""
     cmd += "".join(f" --customise {x}" for x in step_params.get("customise", []))
     customise_commands += step_params.get("customise_commands", [])
     # match central compression on the persisted (final) tier
@@ -165,6 +170,7 @@ def run_step(
     gridpack=None,
     fragment_path=None,
     n_threads=1,
+    pileup_filelist=None,
     verbose=1,
 ):
     """Run one cmsDriver step in its CMSSW env, in work_dir."""
@@ -181,6 +187,7 @@ def run_step(
         gridpack=os.path.abspath(gridpack) if gridpack else None,
         fragment_rel=fragment_rel,
         n_threads=n_threads,
+        pileup_filelist=pileup_filelist,
     )
     from .tools import ps_call
 
@@ -200,6 +207,7 @@ def run_chain(
     fragment_path=None,
     previous_file=None,
     n_threads=1,
+    pileup_filelist=None,
     verbose=1,
 ):
     """Run the linear prod_steps[first_step..last_step] in work_dir, chaining outputs.
@@ -228,6 +236,7 @@ def run_chain(
             gridpack=gridpack,
             fragment_path=fragment_path,
             n_threads=n_threads,
+            pileup_filelist=pileup_filelist,
             verbose=verbose,
         )
         prev = out

dsprod/tasks.py

@@ -126,6 +126,16 @@ def runprod_branches(eras, points):
     return out
 
 
+def premix_dataset_for_era(conditions, era):
+    """The `dbs:` pileup dataset an era's steps resolve, or None when it has no premix step."""
+    for step in conditions[era]["prod_steps"]:
+        params = run_step.resolve_step_params(conditions, era, step)
+        pileup = str(params.get("pileup_input") or "")
+        if pileup.startswith("dbs:") or pileup.startswith("das:"):
+            return pileup[4:]
+    return None
+
+
 def cmssw_releases_for_era(conditions, era):
     """Sorted unique (SCRAM_ARCH, CMSSW) needed to produce `era` (all prod_steps + NANO versions)."""
     releases = set()
@@ -346,6 +356,13 @@ def storage_path(self, *parts):
         """Path of a product relative to `fs_default`: <storage name>/<parts...>."""
         return os.path.join(self.storage_name(), *parts)
 
+    def premix_target(self, era):
+        """Where an era's premix file list lives on `fs_default`.
+
+        Always the production area, also under `--test`: the list depends only on the era.
+        """
+        return self.remote_target(self.prod_storage_name(), "premix", f"{era}.txt")
+
     def gridpack_target(self, gridpack_name):
         """Where a gridpack lives on `fs_default`.
 
@@ -734,6 +751,61 @@ def run(self):
         )
 
 
+class PremixFileList(Task, law.LocalWorkflow):
+    """Resolve an era's premix pileup dataset to a file list, once, and store it on `fs_default`.
+
+    `cmsDriver --pileup_input dbs:<dataset>` resolves the dataset with a DAS query *in the job*.
+    That dataset holds ~38 000 files, and a production asks every job to look it up: at a few
+    thousand concurrent jobs the queries start coming back empty, cmsDriver writes a config with no
+    secondary input, and cmsRun dies with
+
+        NoSecondaryFiles: RootEmbeddedFileSequence no input files specified for secondary input
+
+    after the job has already produced its GEN-SIM. Resolving once here and passing the result as
+    `filelist:` gives every job exactly what a successful DAS query would have, with no per-job
+    lookup. Always local: a worker is the last place that query should run.
+    """
+
+    def create_branch_map(self):
+        eras = [
+            era
+            for era in self.prod_eras
+            if premix_dataset_for_era(self.conditions, era)
+        ]
+        return dict(enumerate(eras))
+
+    def output(self):
+        return self.premix_target(self.branch_data)
+
+    def run(self):
+        era = self.branch_data
+        dataset = premix_dataset_for_era(self.conditions, era)
+        family = self.get_task_family().rsplit(".", 1)[-1]
+        if on_batch_node() and submitted_task_family() not in (None, family):
+            raise RuntimeError(
+                f"the premix file list for {era} is missing from fs_default, and resolving it "
+                f"needs a DAS query that must not run on a worker. Submit {family} (or any task "
+                "that requires it) from a machine with dasgoclient, then resubmit."
+            )
+        _, out, _ = ps_call(
+            [f'dasgoclient -query="file dataset={dataset}"'],
+            shell=True,
+            catch_stdout=True,
+            verbose=1,
+        )
+        files = [
+            line.strip() for line in out.splitlines() if line.strip().endswith(".root")
+        ]
+        if not files:
+            raise RuntimeError(
+                f"DAS returned no files for the premix dataset {dataset}"
+            )
+        print(f"PremixFileList[{era}]: {len(files)} files from {dataset}")
+        with self.output().localize("w") as out_local:
+            with open(out_local.abspath, "w") as f:
+                f.write("\n".join(files) + "\n")
+
+
 class RunProd(Task, HTCondorWorkflow, CrabWorkflow, law.LocalWorkflow):
     """Fused GEN->NANO production for one (era, point, seed); stages one nano per version."""
 
@@ -747,25 +819,35 @@ def create_branch_map(self):
         return dict(enumerate(runprod_branches(self.prod_eras, self.prod_points)))
 
     def workflow_requires(self):
-        return {
+        reqs = {
             "voms": CreateVomsProxy.req(self),
             "cmssw": InstallCMSSW.req(self, workflow="local"),
             "gridpack": MakeGridpack.req(self),
         }
+        if PremixFileList.req(self, workflow="local").get_branch_map():
+            reqs["premix"] = PremixFileList.req(self, workflow="local")
+        return reqs
 
     def requires(self):
         era, pi, _ = self.branch_data
         # MakeGridpack branches over distinct gridpacks, so map this point to its gridpack branch
         gp_branch = self.gridpack_index()[
             self.process.gridpack_name(self.prod_points[pi])
         ]
-        return {
+        reqs = {
             "voms": CreateVomsProxy.req(self),
             "cmssw": InstallCMSSW.req(
                 self, branch=self.prod_eras.index(era), workflow="local"
             ),
             "gridpack": MakeGridpack.req(self, branch=gp_branch),
         }
+        premix = PremixFileList.req(self, workflow="local")
+        eras_with_premix = list(premix.get_branch_map().values())
+        if era in eras_with_premix:
+            reqs["premix"] = PremixFileList.req(
+                self, branch=eras_with_premix.index(era), workflow="local"
+            )
+        return reqs
 
     def _staged_target(self, era, point, version, seed):
         name = self.process.point_name(point)
@@ -790,6 +872,10 @@ def run(self):
                 self.input()["gridpack"].localize("r")
             ).abspath
             work_dir, is_tmp = self.law_job_home()
+            premix = self.input().get("premix")
+            premix_list = (
+                stack.enter_context(premix.localize("r")).abspath if premix else None
+            )
             try:
                 miniaod = run_step.run_chain(
                     self.conditions,
@@ -802,6 +888,7 @@ def run(self):
                     gridpack=gridpack,
                     fragment_path=fragment,
                     n_threads=int(self.n_cpus),
+                    pileup_filelist=premix_list,
                 )
                 for version in self.nano_versions(era):
                     nano_out = run_step.run_nano(