Source code
Revision control
Copy as Markdown
Other Tools
#!/usr/bin/env python3
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
import argparse
import datetime
import json
import os
import sys
import urllib.error
import urllib.request
DEFAULT_CREDENTIALS_FILE = "/builds/worker/gcp-credentials.json"
DEFAULT_OUTPUT_DIR = "/builds/worker/artifacts"
# Only the level-3 mozilla-central cron holds secrets:get on this. Deliberately
# not level-templated: no level-1 name exists, so a try run fails closed.
DEFAULT_SECRET = "project/bhr/aggregation-gcp-key"
# Index route this task publishes to, used to pick up the previous run's
# timeseries state so the roll-up stays incremental.
DEFAULT_STATE_INDEX = "gecko.v2.mozilla-central.latest.firefox.bhr-aggregate"
def _env_float(name, default):
value = os.environ.get(name)
return default if value is None else float(value)
def _env_int(name, default):
value = os.environ.get(name)
return default if value is None else int(value)
def _default_date(offset_days):
return datetime.datetime.now(datetime.UTC).date() - datetime.timedelta(
days=offset_days
)
def _parse_date(value):
if value:
return datetime.date.fromisoformat(value)
return None
def _secret_url(secret_name, with_api_prefix=True):
prefix = "/api" if with_api_prefix else ""
return f"{proxy}{prefix}/secrets/v1/secret/{secret_name}"
def _fetch_secret(secret_name):
errors = []
for with_api_prefix in (True, False):
url = _secret_url(secret_name, with_api_prefix)
try:
with urllib.request.urlopen(url, timeout=60) as response:
return json.loads(response.read().decode("utf-8"))
except (urllib.error.URLError, urllib.error.HTTPError) as error:
errors.append(f"{url}: {error}")
raise RuntimeError("Could not fetch Taskcluster secret:\n" + "\n".join(errors))
def _state_url(index_route, name, with_api_prefix=True):
prefix = "/api" if with_api_prefix else ""
return f"{proxy}{prefix}/index/v1/task/{index_route}/artifacts/public/bhr/{name}"
def _fetch_previous_state(index_route, name, path):
"""Download the last run's timeseries state next to today's artifacts.
Returns True if state was retrieved. A miss is normal and not fatal: the
first run has no predecessor, and build_timeseries refills any window date
absent from state from that day's artifact, so the roll-up self-heals
rather than failing.
"""
errors = []
for with_api_prefix in (True, False):
url = _state_url(index_route, name, with_api_prefix)
try:
with urllib.request.urlopen(url, timeout=300) as response:
data = response.read()
except (urllib.error.URLError, urllib.error.HTTPError) as error:
errors.append(f"{url}: {error}")
continue
with open(path, "wb") as state_file:
state_file.write(data)
print(f"Fetched previous timeseries state ({len(data)} bytes)", flush=True)
return True
print("No previous timeseries state:\n " + "\n ".join(errors), flush=True)
return False
def _write_gcp_credentials(secret_name, secret_key, path):
secret = _fetch_secret(secret_name)["secret"][secret_key]
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as credentials:
if isinstance(secret, str):
credentials.write(secret)
else:
json.dump(secret, credentials)
os.chmod(path, 0o600)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--date", default=os.environ.get("BHR_AGGREGATE_DATE"))
parser.add_argument(
"--date-offset-days",
type=int,
default=_env_int("BHR_AGGREGATE_DATE_OFFSET_DAYS", 4),
)
parser.add_argument(
"--sample-size",
type=float,
default=_env_float("BHR_AGGREGATE_SAMPLE_SIZE", 0.5),
)
parser.add_argument(
"--billing-project",
default=os.environ.get("BHR_AGGREGATE_BILLING_PROJECT", "mozdata"),
)
parser.add_argument(
"--output-dir",
default=os.environ.get("BHR_AGGREGATE_OUTPUT_DIR", DEFAULT_OUTPUT_DIR),
)
parser.add_argument(
"--output-tag",
default=os.environ.get("BHR_AGGREGATE_OUTPUT_TAG", "main"),
)
parser.add_argument(
"--thread-filter",
default=os.environ.get("BHR_AGGREGATE_THREAD_FILTER", "Gecko"),
)
parser.add_argument(
"--credentials-file",
default=os.environ.get(
"GOOGLE_APPLICATION_CREDENTIALS", DEFAULT_CREDENTIALS_FILE
),
)
parser.add_argument(
"--gcp-secret",
default=os.environ.get("BHR_GCP_SECRET", DEFAULT_SECRET),
)
parser.add_argument(
"--gcp-secret-key",
default=os.environ.get("BHR_GCP_SECRET_KEY", "serviceAccount"),
)
parser.add_argument(
"--skip-timeseries",
action="store_true",
default=bool(os.environ.get("BHR_SKIP_TIMESERIES")),
help="Only aggregate the day; do not update the timeseries roll-up.",
)
parser.add_argument(
"--timeseries-window-days",
type=int,
default=_env_int("BHR_TIMESERIES_WINDOW_DAYS", 365),
)
parser.add_argument(
"--timeseries-top-count",
type=int,
default=_env_int("BHR_TIMESERIES_TOP_COUNT", 500),
)
parser.add_argument(
"--timeseries-state-index",
default=os.environ.get("BHR_TIMESERIES_STATE_INDEX", DEFAULT_STATE_INDEX),
)
parser.add_argument(
"topsrcdir",
nargs="?",
default=os.environ.get("GECKO_PATH", os.getcwd()),
)
args = parser.parse_args()
if not 0 < args.sample_size <= 1:
raise ValueError(f"--sample-size must be in (0, 1], got {args.sample_size}")
build_date = _parse_date(args.date) or _default_date(args.date_offset_days)
if not os.path.exists(args.credentials_file):
if not args.gcp_secret:
raise RuntimeError(
"No GCP credentials file or Taskcluster secret configured"
)
_write_gcp_credentials(
args.gcp_secret, args.gcp_secret_key, args.credentials_file
)
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = args.credentials_file
aggregation_dir = os.path.join(
args.topsrcdir,
"toolkit",
"components",
"backgroundhangmonitor",
"aggregation",
)
sys.path.insert(0, aggregation_dir)
import bhr_collection
print(
f"Running BHR aggregation for build date {build_date:%Y-%m-%d} "
f"at sample-size {args.sample_size}",
flush=True,
)
bhr_collection.aggregate(
date=build_date,
sample_size=args.sample_size,
billing_project=args.billing_project,
output_dir=args.output_dir,
output_tag=args.output_tag,
config_overrides={"thread_filter": args.thread_filter},
)
if args.skip_timeseries:
return
import bhr_timeseries
state_name = f"hangs_timeseries_{args.output_tag}_state.json.gz"
_fetch_previous_state(
args.timeseries_state_index,
state_name,
os.path.join(args.output_dir, state_name),
)
# The day just written is the only new input; every earlier day in the
# window is already summarized in the state fetched above.
print(
f"Rolling up the last {args.timeseries_window_days} days",
flush=True,
)
bhr_timeseries.build_timeseries(
input_dir=args.output_dir,
output_dir=args.output_dir,
output_tag=args.output_tag,
end_date=build_date,
window_days=args.timeseries_window_days,
top_count=args.timeseries_top_count,
)
if __name__ == "__main__":
main()