Skip to content

k4bench.provenance.stack

k4bench.provenance.stack

Read an LCG Key4hep nightly's identity and git provenance off CVMFS.

LCG rotates weekday-named views, so their generated timestamp is the release identity. Package revisions come from the manifest's install paths and the .buildinfo_* file installed with each HEAD build. Metadata failures return an empty result rather than failing the benchmark they describe.

StackSnapshot dataclass 

StackSnapshot(packages: dict[str, dict], loaded: frozenset[str] | None = None, datasets: dict[str, str] | None = None, os: str | None = None, complete: bool = False, unmappable: frozenset[str] | None = None)

One recorded stack and the measured scope that supplied it.

RepoRef dataclass 

RepoRef(forge: str, host: str, slug: str)

A package's upstream repository, parsed far enough to build links.

code_search_url 

code_search_url(symbol: str, ref: str) -> str

Where symbol is defined in the repository. GitHub searches only the default branch, and only for a signed-in user; GitLab at ref.

Source code in k4bench/provenance/stack.py 
240
241
242
243
244
245
246
247
def code_search_url(self, symbol: str, ref: str) -> str:
    """Where *symbol* is defined in the repository. GitHub searches only
    the default branch, and only for a signed-in user; GitLab at *ref*."""
    if self.forge == "github":
        query = quote(f"repo:{self.slug} symbol:{symbol}", safe="")
        return f"https://github.com/search?q={query}&type=code"
    return (f"{self.url}/-/search?scope=blobs&search={quote(symbol, safe='')}"
            f"&repository_ref={quote(ref, safe='')}")

capture_stack 

capture_stack(run_dir: Path, info: dict, env: dict[str, str]) -> None

Save the manifests and reduce all measured configs' private path lists.

Missing instrumentation on any measured config makes the union unknown. Dataset directories come from the instrumented process's environment; the job environment also records them when instrumentation is unavailable.

Source code in k4bench/provenance/stack.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def capture_stack(run_dir: Path, info: dict, env: dict[str, str]) -> None:
    """Save the manifests and reduce all measured configs' private path lists.

    Missing instrumentation on any measured config makes the union unknown.
    Dataset directories come from the instrumented process's environment;
    the job environment also records them when instrumentation is unavailable.
    """
    info["g4_datasets"] = {
        key: Path(value).name for key, value in sorted(env.items())
        if key.startswith("G4") and key.endswith("DATA") and value
    }
    handoffs = [loaded_paths_file(run_dir, label) for label in info.get("configs", [])]
    try:
        manifest = _manifest(Path(env["K4H_STACK_SETUP"]).resolve())
        if manifest is None:
            return
        sources = [manifest, manifest.with_name(manifest.name.replace("externals", "contrib"))]
        packages = {}
        for source, filename in zip(sources, STACK_MANIFESTS):
            raw = source.read_bytes()
            (run_dir / filename).write_bytes(gzip.compress(raw, mtime=0))
            packages.update(parse_manifest(raw.decode()))
        info["k4h_stack_capture"] = 1
        paths, datasets = [], None
        for handoff in handoffs:
            rows = handoff.read_text().splitlines()
            if not rows or rows[0] != "# k4bench loaded paths v1":
                return
            paths.extend(re.sub(r"\\([0-7]{3})", lambda m: chr(int(m[1], 8)), row)
                         for row in rows[1:] if row.startswith("/"))
            config_data = {key: Path(value).name for row in rows[1:]
                           if row.startswith("G4") and "=" in row
                           for key, value in [row.split("=", 1)]}
            if datasets is not None and config_data != datasets:
                # The job environment would now name data some config did not use.
                _log.warning("Geant4 dataset environment differs across configs")
                info.pop("g4_datasets", None)
                return
            datasets = config_data
        if not handoffs or not paths:
            return
        # Gaudi executes these options before enabling the auditor. Staged
        # copies have local paths, so retain the source paths the job recorded.
        if info.get("tool") == "k4run":
            stage = Path(info.get("resolved_stage_dir") or ".")
            paths.extend(str(stage / option) for option in info.get("options", []))
        # ddsim executes its steering file without mapping it, and the file can
        # belong to a Python-only package such as CLDConfig.
        if steering := info.get("resolved_steering_file"):
            paths.append(steering)
        loaded, unknown = mapped_packages(paths, packages)
        unmatched = [path for path in unknown if path.startswith("/cvmfs/")]
        if unmatched:
            _log.warning("Mapped CVMFS paths absent from manifests: %s", unmatched)
        info["k4h_loaded_packages"] = {
            name: {"version": packages[name]["version"], **build_info(packages[name]["install_path"]),
                   **resolved_install(packages[name]["install_path"]),
                   **info.get("k4h_packages", {}).get(name, {})}
            for name in sorted(loaded)
        }
        # k4run appends its imported Python modules; ddsim records only mapped
        # files, so a HEAD package without a shared library is never seen even
        # when the steering file imports it. Blame keeps those as candidates.
        info["k4h_unmappable_packages"] = [] if info.get("tool") == "k4run" else sorted(
            name for name, package in packages.items()
            if package["version"] == "HEAD" and name not in loaded
            and not any(Path(package["install_path"]).glob("lib*/**/*.so*"))
        )
        info["g4_datasets"] = datasets
    except Exception as exc:
        _log.warning("Loaded stack provenance not recorded: %s", exc)
    finally:
        for path in handoffs:
            try:
                path.unlink(missing_ok=True)
            except OSError:
                _log.warning("Could not remove private provenance handoff %s", path)

parse_manifest 

parse_manifest(source: str, *, contrib_root: Path = CONTRIB_ROOT) -> dict[str, dict]

Read package rows, ignoring the manifest's platform/compiler headers.

Source code in k4bench/provenance/stack.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def parse_manifest(source: str, *, contrib_root: Path = CONTRIB_ROOT) -> dict[str, dict]:
    """Read package rows, ignoring the manifest's platform/compiler headers."""
    packages = {}
    for row in source.splitlines():
        fields = [field.strip() for field in row.split(";")]
        if len(fields) < 4 or not fields[0] or not fields[3]:
            continue
        name, build_hash, version, install = fields[:4]
        path = Path(install)
        packages[name] = {
            "version": version, "hash": build_hash,
            "install_path": str(path if path.is_absolute() else contrib_root / path),
        }
    return packages

build_info 

build_info(install: str) -> dict[str, str]

Available upstream and toolchain revisions of one installed build.

Source code in k4bench/provenance/stack.py 
135
136
137
138
139
140
141
142
143
144
145
def build_info(install: str) -> dict[str, str]:
    """Available upstream and toolchain revisions of one installed build."""
    try:
        source = next(Path(install).glob(".buildinfo_*.txt")).read_text()
    except (OSError, StopIteration):
        return {}
    return {
        key: match.group(1)
        for key, pattern in (("commit", _REVISION_RE), ("githash", _GITHASH_RE))
        if (match := pattern.search(source))
    }

resolved_install 

resolved_install(install: str) -> dict[str, str]

The directory install resolves to now, where symlinks lead elsewhere.

Recorded at capture: the auditor names libraries by their resolved paths, and LCG's rotating names may point elsewhere when the run is read.

Source code in k4bench/provenance/stack.py 
148
149
150
151
152
153
154
155
def resolved_install(install: str) -> dict[str, str]:
    """The directory *install* resolves to now, where symlinks lead elsewhere.

    Recorded at capture: the auditor names libraries by their resolved paths,
    and LCG's rotating names may point elsewhere when the run is read.
    """
    real = str(Path(install).resolve())
    return {"install_realpath": real} if real != install.rstrip("/") else {}

mapped_packages 

mapped_packages(paths: list[str], packages: dict[str, dict]) -> tuple[set[str], list[str]]

Resolve view symlinks and match the longest manifest install prefix.

Unmatched paths are returned to the caller, including OS and harness files. No CVMFS location is assumed: contrib and stable releases use the same match.

Source code in k4bench/provenance/stack.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def mapped_packages(paths: list[str], packages: dict[str, dict]) -> tuple[set[str], list[str]]:
    """Resolve view symlinks and match the longest manifest install prefix.

    Unmatched paths are returned to the caller, including OS and harness files.
    No CVMFS location is assumed: contrib and stable releases use the same match.
    """
    prefixes = sorted(
        ((str(Path(p["install_path"]).resolve()).rstrip("/") + "/", name)
         for name, p in packages.items()),
        key=lambda item: -len(item[0]),
    )
    loaded, unknown = set(), []
    for path in sorted(set(paths)):
        real = str(Path(path.removesuffix(" (deleted)")).resolve())
        name = next((name for prefix, name in prefixes if real.startswith(prefix)), None)
        if name is None:
            unknown.append(path)
        else:
            loaded.add(name)
    return loaded, unknown

stack_snapshot 

stack_snapshot(info: dict, machine: dict, manifests: list[str] | tuple[str, ...] = ()) -> StackSnapshot

Combine immutable artifacts without resolving their now-rotating paths.

Source code in k4bench/provenance/stack.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def stack_snapshot(info: dict, machine: dict, manifests: list[str] | tuple[str, ...] = ()) -> StackSnapshot:
    """Combine immutable artifacts without resolving their now-rotating paths."""
    packages = {}
    for source in manifests:
        packages.update(parse_manifest(source))
    for name, package in info.get("k4h_packages", {}).items():
        packages.setdefault(name, {}).update(package)
    loaded = info.get("k4h_loaded_packages")
    for name, package in (loaded or {}).items():
        packages.setdefault(name, {}).update(package)
    unmappable = info.get("k4h_unmappable_packages")
    return StackSnapshot(
        packages, frozenset(loaded) if isinstance(loaded, dict) else None,
        info.get("g4_datasets"), machine.get("os") or None, len(manifests) == 2,
        frozenset(unmappable) if isinstance(unmappable, list) else None,
    )

stack_identity 

stack_identity(stack_setup: str | Path) -> tuple[str, str]

Return (publication_date, platform) from an LCG view setup.

Source code in k4bench/provenance/stack.py 
210
211
212
213
214
215
216
217
218
219
def stack_identity(stack_setup: str | Path) -> tuple[str, str]:
    """Return ``(publication_date, platform)`` from an LCG view setup."""
    path = Path(stack_setup).resolve()
    try:
        match = _GENERATED_RE.search(path.read_text())
        if match:
            return parsedate_to_datetime(match.group(1)).date().isoformat(), path.parent.name
    except (OSError, TypeError, ValueError):
        pass
    return "", "unknown"

parse_repo 

parse_repo(url: str | None) -> RepoRef | None

Parse a GitHub/GitLab repository URL used by attribution links.

Source code in k4bench/provenance/stack.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def parse_repo(url: str | None) -> RepoRef | None:
    """Parse a GitHub/GitLab repository URL used by attribution links."""
    match = _URL_RE.match(url.strip()) if url else None
    if not match:
        return None
    host, slug = match.group("host"), match.group("slug")
    if host == "github.com":
        if slug.count("/") != 1:
            return None
        forge = "github"
    elif "gitlab" in host:
        forge = "gitlab"
    else:
        return None
    return RepoRef(forge, host, slug)

read_stack 

read_stack(stack_setup: str | Path) -> tuple[Path | None, dict[str, dict]]

Return the LCG manifest and git-built HEAD packages for a view setup.

Source code in k4bench/provenance/stack.py 
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
def read_stack(stack_setup: str | Path) -> tuple[Path | None, dict[str, dict]]:
    """Return the LCG manifest and git-built HEAD packages for a view setup."""
    setup = Path(stack_setup).resolve()
    manifest = _manifest(setup)
    if manifest is None:
        return None, {}
    try:
        rows = manifest.read_text().splitlines()
    except OSError as exc:
        _log.warning("cannot read LCG manifest '%s' (%s)", manifest, exc)
        return None, {}

    packages: dict[str, dict] = {}
    build_revisions: dict[str, str] = {}
    for name, package in parse_manifest("\n".join(rows)).items():
        if package["version"] != "HEAD":
            continue
        metadata = build_info(package["install_path"])
        if metadata.get("commit"):
            build_revisions[name] = metadata.get("githash", "")
            packages[name] = {
                "commit": metadata["commit"], "version": "HEAD", "repo_url": None,
            }
    # Incremental builds can use different toolchains within one view. Resolve
    # each package against its own build, fetching each distinct revision once.
    repos_by_revision = {
        revision: _repository_urls(setup.parents[2].name, revision)
        for revision in dict.fromkeys(build_revisions.values()) if revision
    }
    for name, package in packages.items():
        package["repo_url"] = repos_by_revision.get(build_revisions[name], {}).get(name.lower())
    return manifest, packages