diff --git a/.gitignore b/.gitignore index 364fcb8..38c34a0 100644 --- a/.gitignore +++ b/.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 diff --git a/docs/concepts/backends.md b/docs/concepts/backends.md index 89f71d4..2d5da59 100644 --- a/docs/concepts/backends.md +++ b/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.'` 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` diff --git a/docs/concepts/tasks.md b/docs/concepts/tasks.md index d95438e..7387e9e 100644 --- a/docs/concepts/tasks.md +++ b/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` +(`/premix/.txt`), once, and `RunProd` passes it to `cmsDriver` as +`--pileup_input filelist:...`. + +Without it, `--pileup_input dbs:` 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 `_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 diff --git a/dsprod/crab.py b/dsprod/crab.py index a117893..bd97896 100644 --- a/dsprod/crab.py +++ b/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: diff --git a/dsprod/run_step.py b/dsprod/run_step.py index 0dcb7f8..070bdcb 100644 --- a/dsprod/run_step.py +++ b/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 diff --git a/dsprod/tasks.py b/dsprod/tasks.py index d7b4d9b..8b39f97 100644 --- a/dsprod/tasks.py +++ b/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`: /.""" 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:` 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,11 +819,14 @@ 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 @@ -759,13 +834,20 @@ def requires(self): 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( diff --git a/env.sh b/env.sh index b0692b6..fe731e0 100644 --- a/env.sh +++ b/env.sh @@ -156,8 +156,26 @@ action() { cat > "$ANALYSIS_PATH/soft/bin/crab" <<'CRABWRAP' #!/bin/bash source /cvmfs/cms.cern.ch/cmsset_default.sh +# CRAB rewrites its task cache ~/.crab3 (via ~/.crab3.) on *every* command, status queries +# included -- so with $HOME on AFS a long production dies the moment the AFS token lapses: +# PermissionError: [Errno 13] Permission denied: '/afs/cern.ch/user/x/xyz/.crab3.' +# and law reports it as a status-query failure for every job at once. DSProd keeps everything else +# off AFS, so give CRAB a home of its own too. law passes --proxy to submit/status/kill, so it +# never needs ~/.globus from the real home. +export HOME="${DSPROD_CRAB_HOME:-${TMPDIR:-/tmp}/dsprod_crab_home_$(id -u)}" +mkdir -p "$HOME" || exit 1 _c=$(ls -d "$ANALYSIS_PATH"/soft/CMSSW_*/ 2>/dev/null | sort | tail -1) [ -n "$_c" ] && { cd "$_c/src" && eval $(scramv1 runtime -sh 2>/dev/null); cd - >/dev/null; } +# crab drops a crab.log wherever it is run from, and law calls status/kill without setting a +# directory, so they inherited the caller's cwd -- the production area. Run those from crab's own +# home. `submit` must keep its directory: law runs it with cwd set to the job-file directory and +# the generated config names `scriptExe` and `inputFiles` relative to it, which CRAB resolves +# against the cwd ("Cannot find the file crab_wrapper_*.sh specified in the JobType.scriptExe +# configuration parameter"). Its log then stays next to the job files, under data/jobs/. +case "$1" in + submit) ;; + *) cd "$HOME" || exit 1 ;; +esac exec /cvmfs/cms.cern.ch/common/crab "$@" CRABWRAP chmod +x "$ANALYSIS_PATH/soft/bin/crab"