Skip to content

Commit 2034a92

Browse files
authored
getCCDBTimeMachineTimestamp.py: publish the NotAfter estimate to CCDB (#2509)
This allows to store the estimated NotAfter timestamp of a period and pass in CCDB as a `CCDBTimeMachineTS` object, so that the estimate is done once per pass and not in every job. - The script first queries CCDB for an existing object for the Period and Pass, and only estimates from the catalogue when none is found. - `--publish-to-ccdb` uploads the estimate, `--force-overwrite` replaces an existing object and `--skip-ccdb-query` skips the lookup. - The CCDB cross-check now uses the run start timestamp. - Only the value goes to stdout; all log lines go to stderr. Nothing uses the script in production yet. https://its.cern.ch/jira/browse/O2-6543
1 parent 43060e0 commit 2034a92

1 file changed

Lines changed: 224 additions & 44 deletions

File tree

‎GRID/utils/getCCDBTimeMachineTimestamp.py‎

Lines changed: 224 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -13,17 +13,40 @@
1313
# started 25.08.2026; Sandro Wenzel
1414

1515
import argparse
16+
import dataclasses
17+
from dataclasses import dataclass, asdict, fields
1618
import json
1719
import os
20+
import requests
1821
import statistics
1922
import subprocess
2023
import sys
2124
import urllib.request
2225
from datetime import datetime, timezone
2326

27+
# hack to find the script with helper functions
28+
o2dpg_root = os.environ.get("O2DPG_ROOT")
29+
if o2dpg_root is None:
30+
raise EnvironmentError("O2DPG_ROOT is not set in the environment.")
31+
mc_prodinfo_path = os.path.abspath(os.path.join(o2dpg_root, "MC", "prodinfo"))
32+
sys.path.append(mc_prodinfo_path)
33+
# FIXME: these should probably be moved to a common script
34+
from mcprodinfo_ccdb_upload import extract_metadata_blocks_from_CCDB
35+
2436
CCDB_HOST = "http://alice-ccdb.cern.ch"
2537

2638

39+
@dataclass(frozen=True)
40+
class CCDBTimeMachineTS:
41+
"""
42+
struct for CCDB time machine timestamp
43+
"""
44+
LPMProductionTag: str
45+
Period: str
46+
Pass: str
47+
NotAfter: int
48+
49+
2750
def alien_json(*args):
2851
"""Run alien.py with -json and return the parsed 'results' list."""
2952
cmd = ["alien.py", "-json"] + list(args)
@@ -68,11 +91,7 @@ def human(ts_ms):
6891
return datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
6992

7093

71-
def ccdb_cross_check(path, not_after, timeout=15):
72-
"""Report which object a correctly time-machine-pinned query at not_after
73-
would actually resolve to, and flag if CCDB has since been updated for the
74-
same validity window. Never raises; returns None on any failure."""
75-
url = f"{CCDB_HOST}/browse/{path}/{not_after}"
94+
def get_ccdb_objects(url, timeout = 15):
7695
req = urllib.request.Request(url, headers={"Accept": "application/json"})
7796
try:
7897
with urllib.request.urlopen(req, timeout=timeout) as resp:
@@ -81,55 +100,72 @@ def ccdb_cross_check(path, not_after, timeout=15):
81100
return {"error": str(exc)}
82101

83102
objects = data.get("objects", [])
103+
return objects
104+
105+
106+
def get_run_start_tf(run):
107+
run_info_url = f"{CCDB_HOST}/browse/RCT/Info/RunInformation/{run}"
108+
objects = get_ccdb_objects(run_info_url)
109+
110+
return objects[0]["STF"]
111+
112+
113+
def ccdb_cross_check(run, path, not_after, timeout=15):
114+
"""Report which object a correctly time-machine-pinned query at not_after
115+
would actually resolve to, and flag if CCDB has since been updated for the
116+
same validity window. Never raises; returns None on any failure."""
117+
timestamp = get_run_start_tf(run)
118+
url = f"{CCDB_HOST}/browse/{path}/{timestamp}"
119+
result = {"timestamp": timestamp}
120+
121+
objects = get_ccdb_objects(url)
84122
if not objects:
85-
return {"error": "no CCDB object covers this instant"}
123+
return {**result, "error": "no CCDB object covers this instant"}
86124

87125
as_of = [o for o in objects if o.get("Created", 0) <= not_after]
88126
newest = max(objects, key=lambda o: o.get("Created", 0))
89-
pinned = max(as_of, key=lambda o: o.get("Created", 0)) if as_of else None
90-
outdated = pinned is not None and newest["Created"] > pinned["Created"]
91-
return {"newest": newest, "as_of": pinned, "outdated": outdated}
127+
result.update({"newest": newest, "as_of": None, "outdated": False})
128+
if as_of:
129+
result["as_of"] = max(as_of, key=lambda o: o.get("Created", 0))
130+
result["outdated"] = newest["Created"] > result["as_of"]["Created"]
131+
return result
92132

93133

94134
def describe_object(o):
95135
return (f"Created {human(o['Created'])}, JIRA={o.get('JIRA', '?')}, "
96136
f"comment=\"{o.get('comment', '')}\"")
97137

98138

99-
def main():
100-
parser = argparse.ArgumentParser(
101-
description="Estimate a CCDB TimeMachine not-after timestamp for a run+pass, "
102-
"from AliEn catalog timestamps of that pass's own reco output.")
103-
parser.add_argument("--year", default=os.environ.get("ALIEN_JDL_LPMANCHORYEAR"),
104-
help="data-taking year, e.g. 2023 (default: $ALIEN_JDL_LPMANCHORYEAR)")
105-
parser.add_argument("--period", default=os.environ.get("ALIEN_JDL_LPMANCHORPRODUCTION"),
106-
help="period name, e.g. LHC23s (default: $ALIEN_JDL_LPMANCHORPRODUCTION)")
107-
parser.add_argument("--run", default=os.environ.get("ALIEN_JDL_LPMANCHORRUN"),
108-
help="run number (default: $ALIEN_JDL_LPMANCHORRUN)")
109-
parser.add_argument("--pass", dest="passname", default=os.environ.get("ALIEN_JDL_LPMANCHORPASSNAME"),
110-
help="pass name, e.g. apass4 (default: $ALIEN_JDL_LPMANCHORPASSNAME)")
111-
parser.add_argument("--samples", type=int, default=9,
112-
help="number of per-timeframe output files to sample (default: 9)")
113-
parser.add_argument("--ccdb-check-path", default="TPC/Calib/CorrectionMapV2",
114-
help="CCDB path to cross-check the estimate against (default: TPC/Calib/CorrectionMapV2)")
115-
parser.add_argument("--no-ccdb-check", action="store_true",
116-
help="skip the CCDB cross-check")
117-
args = parser.parse_args()
139+
def make_cert_paths(cert_dir="/tmp"):
140+
key_path = os.environ.get("JALIEN_TOKEN_KEY")
141+
cert_path = os.environ.get("JALIEN_TOKEN_CERT")
142+
if key_path == None and cert_path == None:
143+
uid = os.getuid()
144+
cert_path = os.path.join(cert_dir, f"tokencert_{uid}.pem")
145+
key_path = os.path.join(cert_dir, f"tokenkey_{uid}.pem")
146+
147+
return cert_path, key_path
118148

119-
missing = [n for n in ("year", "period", "run", "passname") if not getattr(args, n)]
120-
if missing:
121-
parser.error("missing required value(s): " + ", ".join(missing))
122149

123-
pass_dir = f"/alice/data/{args.year}/{args.period}/{args.run}/{args.passname}/"
150+
def get_NotAfter(year, period, run, passname, samples=9, ccdb_check_path="TPC/Calib/CorrectionMapV2", ccdb_check=True):
151+
"""
152+
Estimates the CCDBTimeMachine NotAfter timestamp from the output of the production of a given period/run/pass.
153+
Cross-checks the estimated NotAfter timestamp vs the creation time of a CCDB object which is currently
154+
valid for the given run.
155+
156+
Returns:
157+
int: CCDBTimeMachineTS NotAfter timestamp
158+
"""
159+
pass_dir = f"/alice/data/{year}/{period}/{run}/{passname}/"
124160
print(f"[getCCDBTimeMachineTimestamp] Looking for TF output under {pass_dir}", file=sys.stderr)
125161

126-
matches = find_tf_output_files(pass_dir, args.run)
162+
matches = find_tf_output_files(pass_dir, run)
127163
if not matches:
128-
sys.exit(f"[getCCDBTimeMachineTimestamp] No o2_ctf_run{int(args.run):08d}_orbit* output found "
129-
f"under {pass_dir}. Check year/period/run/pass, or this run+pass may not use this "
130-
f"catalog layout. Refusing to guess a timestamp.")
164+
sys.exit(f"[getCCDBTimeMachineTimestamp] No o2_ctf_run{int(run):08d}_orbit* output found "
165+
f"under {pass_dir}. Check year/period/run/pass, or this run+pass may not use this "
166+
f"catalog layout. Refusing to guess a timestamp.")
131167

132-
sampled = sample_evenly(matches, args.samples)
168+
sampled = sample_evenly(matches, samples)
133169
ctimes = []
134170
for lfn in sampled:
135171
ctime = stat_ctime(lfn)
@@ -139,33 +175,177 @@ def main():
139175

140176
if len(ctimes) < 2:
141177
sys.exit(f"[getCCDBTimeMachineTimestamp] Only got {len(ctimes)} usable timestamp(s) out of "
142-
f"{len(sampled)} sampled files ({len(matches)} found in total); refusing to guess.")
178+
f"{len(sampled)} sampled files ({len(matches)} found in total); refusing to guess.")
143179

144180
not_after = int(statistics.median(ctimes))
145181
spread_h = (max(ctimes) - min(ctimes)) / 3.6e6
146182
print(f"[getCCDBTimeMachineTimestamp] {len(ctimes)} samples out of {len(matches)} TF outputs found; "
147183
f"span {human(min(ctimes))} .. {human(max(ctimes))} ({spread_h:.1f} h)", file=sys.stderr)
148-
print(f"[getCCDBTimeMachineTimestamp] Recommended not-after: {not_after} ({human(not_after)})",
184+
print(f"[getCCDBTimeMachineTimestamp] Recommended NotAfter: {not_after} ({human(not_after)})",
149185
file=sys.stderr)
150186

151-
if not args.no_ccdb_check:
152-
check = ccdb_cross_check(args.ccdb_check_path, not_after)
187+
if ccdb_check:
188+
check = ccdb_cross_check(run, ccdb_check_path, not_after)
189+
check_timestamp = f"/{check['timestamp']}" if check else ""
153190
if check is None or "error" in check:
154191
reason = check["error"] if check else "unknown error"
155-
print(f"[getCCDBTimeMachineTimestamp] CCDB cross-check on {args.ccdb_check_path} skipped: {reason}",
192+
print(f"[getCCDBTimeMachineTimestamp] CCDB cross-check on {ccdb_check_path}{check_timestamp} skipped: {reason}",
156193
file=sys.stderr)
157194
elif check["as_of"] is None:
158-
print(f"[getCCDBTimeMachineTimestamp] CCDB cross-check on {args.ccdb_check_path}: no object "
195+
print(f"[getCCDBTimeMachineTimestamp] CCDB cross-check on {ccdb_check_path}{check_timestamp}: no object "
159196
f"existed yet at the recommended timestamp (earliest is {describe_object(check['newest'])})",
160197
file=sys.stderr)
161198
else:
162-
print(f"[getCCDBTimeMachineTimestamp] CCDB cross-check on {args.ccdb_check_path}: as-of object is "
199+
print(f"[getCCDBTimeMachineTimestamp] CCDB cross-check on {ccdb_check_path}{check_timestamp}: as-of object is "
163200
f"{describe_object(check['as_of'])}", file=sys.stderr)
164201
if check["outdated"]:
165202
print(f"[getCCDBTimeMachineTimestamp] note: CCDB's current object for this window is newer "
166203
f"({describe_object(check['newest'])}) - it postdates this pass's processing and was "
167204
f"not what this run actually saw.", file=sys.stderr)
168205

206+
return not_after
207+
208+
209+
def query_CCDBTimeMachineTS(base_url, user, period, passname):
210+
"""
211+
Queries CCDBTimeMachineTS from CCDB for a given period and pass.
212+
213+
Returns:
214+
obtained NotAfter timestamp (int) or None
215+
"""
216+
# Construct user ID-specific cert and key paths
217+
cert_path, key_path = make_cert_paths()
218+
219+
# Build full URL
220+
query = f"1/Period={period}/Pass={passname}"
221+
user_path = 'Users/' + user[0] + '/' + user
222+
url = f"{base_url}/browse/{user_path}/CCDBTimeMachineTS/{query}"
223+
224+
print(f"[getCCDBTimeMachineTimestamp] Checking if CCDBTimeMachineTS already exists: {url}", file=sys.stderr)
225+
226+
response = requests.get(url, cert=(cert_path, key_path), verify=False)
227+
if response.status_code == 200:
228+
meta = extract_metadata_blocks_from_CCDB(response.content.decode('utf-8'))
229+
if (len(meta) > 0):
230+
def filter_known_fields(cls, data: dict) -> dict:
231+
valid_keys = {f.name for f in fields(cls)}
232+
return {k: v for k, v in data.items() if k in valid_keys}
233+
234+
clean_meta = filter_known_fields(CCDBTimeMachineTS, meta[0])
235+
print(f"[getCCDBTimeMachineTimestamp] Found object for {clean_meta['Period']}, Pass: {clean_meta['Pass']}, "
236+
f"NotAfter: {clean_meta['NotAfter']} ({human(clean_meta['NotAfter'])}), uploaded by LPMProductionTag {clean_meta['LPMProductionTag']}",
237+
file=sys.stderr)
238+
return clean_meta['NotAfter']
239+
240+
return None
241+
242+
243+
def upload_CCDBTimeMachineTS(base_url, user, keys):
244+
"""
245+
Uploads an empty .dat file as CCDBTimeMachineTS with the corresponding meta data provided by `keys`
246+
247+
Returns:
248+
Response object from the POST request
249+
"""
250+
# Create an empty file
251+
empty_file = "empty.dat"
252+
with open(empty_file, "w") as f:
253+
f.write("0")
254+
255+
# Construct user ID-specific cert and key paths
256+
cert_path, key_path = make_cert_paths()
257+
258+
# Build full URL
259+
query = "/".join(f"{k}={v}" for k, v in keys.items())
260+
user_path = 'Users/' + user[0] + '/' + user
261+
start = 1
262+
stop = start + 1
263+
url = f"{base_url}/{user_path}/CCDBTimeMachineTS/{start}/{stop}/{query}"
264+
265+
print(f"[getCCDBTimeMachineTimestamp] Uploading CCDBTimeMachineTS: {url}", file=sys.stderr)
266+
267+
# Prepare request
268+
with open(empty_file, 'rb') as f:
269+
files = {'blob': f}
270+
response = requests.post(url, files=files, cert=(cert_path, key_path), verify=False)
271+
272+
os.remove(empty_file)
273+
274+
if response.status_code != 201:
275+
print(f"[getCCDBTimeMachineTimestamp] Upload failed, response: {response.status_code}", file=sys.stderr)
276+
277+
278+
def get_CCDBTimeMachineTS(production_tag, year, period, run, passname,
279+
samples=9, ccdb_check_path="TPC/Calib/CorrectionMapV2", ccdb_check=True,
280+
ccdb_url="https://alice-ccdb.cern.ch", username="aliprod",
281+
force_overwrite=False, publish_to_ccdb=True, skip_ccdb_query=False):
282+
"""
283+
Makes a query to CCDB for a CCDBTimeMachineTS for given period and pass.
284+
If an object is found, the NotAfter timestamp is returned.
285+
If no object is found, the NotAfter timestamp is estimated from the reco output of the given period/run/pass.
286+
In case an object is found, a new upload can be forced by setting the argument `force_overwrite=True`
287+
288+
Returns:
289+
int: Either the queried or the estimated NotAfter timestamp
290+
"""
291+
if not skip_ccdb_query:
292+
not_after = query_CCDBTimeMachineTS(ccdb_url, username, period, passname)
293+
else:
294+
not_after = None
295+
296+
if not_after is not None and force_overwrite == True:
297+
print("[getCCDBTimeMachineTimestamp] Existing object will be overwritten", file=sys.stderr)
298+
299+
if not_after == None or force_overwrite == True:
300+
print("[getCCDBTimeMachineTimestamp] Estimating CCDBTimeMachineTS for "
301+
f"LPMProductionTag: {production_tag}, Period: {period}, Pass: {passname}", file=sys.stderr)
302+
not_after = get_NotAfter(year, period, run, passname, samples, ccdb_check_path, ccdb_check)
303+
time_machine_ts = CCDBTimeMachineTS(production_tag, period, passname, not_after)
304+
if publish_to_ccdb:
305+
upload_CCDBTimeMachineTS(ccdb_url, username, dataclasses.asdict(time_machine_ts))
306+
307+
return not_after
308+
309+
310+
def main():
311+
parser = argparse.ArgumentParser(
312+
description="Estimate a CCDB TimeMachine not-after timestamp for a run+pass, "
313+
"from AliEn catalog timestamps of that pass's own reco output.")
314+
parser.add_argument("--production-tag", default=os.environ.get("ALIEN_JDL_LPMPRODUCTIONTAG"),
315+
help="production tag to mark which production uploads an object (default: $ALIEN_JDL_LPMPRODUCTIONTAG)")
316+
parser.add_argument("--year", default=os.environ.get("ALIEN_JDL_LPMANCHORYEAR"),
317+
help="data-taking year, e.g. 2023 (default: $ALIEN_JDL_LPMANCHORYEAR)")
318+
parser.add_argument("--period", default=os.environ.get("ALIEN_JDL_LPMANCHORPRODUCTION"),
319+
help="period name, e.g. LHC23s (default: $ALIEN_JDL_LPMANCHORPRODUCTION)")
320+
parser.add_argument("--run", default=os.environ.get("ALIEN_JDL_LPMANCHORRUN"),
321+
help="run number (default: $ALIEN_JDL_LPMANCHORRUN)")
322+
parser.add_argument("--pass", dest="passname", default=os.environ.get("ALIEN_JDL_LPMANCHORPASSNAME"),
323+
help="pass name, e.g. apass4 (default: $ALIEN_JDL_LPMANCHORPASSNAME)")
324+
parser.add_argument("--samples", type=int, default=9,
325+
help="number of per-timeframe output files to sample (default: 9)")
326+
parser.add_argument("--ccdb-check-path", default="TPC/Calib/CorrectionMapV2",
327+
help="CCDB path to cross-check the estimate against (default: TPC/Calib/CorrectionMapV2)")
328+
parser.add_argument("--no-ccdb-check", action="store_true",
329+
help="skip the CCDB cross-check after calculating the NotAfter")
330+
parser.add_argument("--publish-to-ccdb", action="store_true",
331+
help="upload the CCDBTimeMachineTS object to the CCDB")
332+
parser.add_argument("--skip-ccdb-query", action="store_true",
333+
help="skip CCDB query for existing objects while publishing")
334+
parser.add_argument("--force-overwrite", action="store_true",
335+
help="overwrite the current CCDBTimeMachineTS object in the CCDB if one exists")
336+
parser.add_argument("--ccdb-url", default="https://alice-ccdb.cern.ch",
337+
help="ccdb host for the upload")
338+
parser.add_argument("--username", default="aliprod",
339+
help="username to be used for the upload to CCDB")
340+
args = parser.parse_args()
341+
342+
missing = [n for n in ("year", "period", "run", "passname") if not getattr(args, n)]
343+
if missing:
344+
parser.error("missing required value(s): " + ", ".join(missing))
345+
346+
not_after = get_CCDBTimeMachineTS(args.production_tag, args.year, args.period, args.run, args.passname,
347+
args.samples, args.ccdb_check_path, not args.no_ccdb_check, args.ccdb_url, args.username,
348+
args.force_overwrite, args.publish_to_ccdb, args.skip_ccdb_query)
169349
print(not_after)
170350

171351

0 commit comments

Comments
 (0)