Source code
Revision control
Copy as Markdown
Other Tools
# 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 json
import os
import sys
import traceback
import six
from mach.util import get_state_dir
from mozbuild.base import MozbuildObject
from mozversioncontrol import MissingVCSExtension, get_repository_object
from .lando import push_to_lando_try
from .util.estimates import duration_summary
from .util.manage_estimates import (
download_task_history_data,
make_trimmed_taskgraph_cache,
)
GIT_CINNABAR_NOT_FOUND = """
Could not detect `git-cinnabar`.
The `mach try` command requires git-cinnabar to be installed when
pushing from git. Please install it by running:
$ ./mach vcs-setup
""".lstrip()
HG_PUSH_TO_TRY_NOT_FOUND = """
Could not detect `push-to-try`.
The `mach try` command requires the push-to-try extension enabled
when pushing from hg. Please install it by running:
$ ./mach vcs-setup
""".lstrip()
VCS_NOT_FOUND = """
Could not detect version control. Only `hg` or `git` are supported.
""".strip()
UNCOMMITTED_CHANGES = """
ERROR please commit changes before continuing
""".strip()
MAX_HISTORY = 10
here = os.path.abspath(os.path.dirname(__file__))
build = MozbuildObject.from_environment(cwd=here)
vcs = get_repository_object(build.topsrcdir)
history_path = os.path.join(
get_state_dir(specific_to_topsrcdir=True), "history", "try_task_configs.json"
)
def write_task_config(try_task_config):
config_path = os.path.join(vcs.path, "try_task_config.json")
with open(config_path, "w") as fh:
json.dump(try_task_config, fh, indent=4, separators=(",", ": "), sort_keys=True)
fh.write("\n")
return config_path
def write_task_config_history(msg, try_task_config):
if not os.path.isfile(history_path):
if not os.path.isdir(os.path.dirname(history_path)):
os.makedirs(os.path.dirname(history_path))
history = []
else:
with open(history_path) as fh:
history = fh.read().strip().splitlines()
history.insert(0, json.dumps([msg, try_task_config]))
history = history[:MAX_HISTORY]
with open(history_path, "w") as fh:
fh.write("\n".join(history))
def check_working_directory(push=True):
if not push:
return
if not vcs.working_directory_clean():
print(UNCOMMITTED_CHANGES)
sys.exit(1)
def generate_try_task_config(method, labels, params=None, routes=None):
params = params or {}
# The user has explicitly requested a set of jobs, so run them all
# regardless of optimization (unless the selector explicitly sets this to
# True). Their dependencies can be optimized though.
params.setdefault("optimize_target_tasks", False)
# Remove selected labels from 'existing_tasks' parameter if present
if "existing_tasks" in params:
params["existing_tasks"] = {
label: tid
for label, tid in params["existing_tasks"].items()
if label not in labels
}
try_config = params.setdefault("try_task_config", {})
try_config.setdefault("env", {})["TRY_SELECTOR"] = method
try_config["tasks"] = sorted(labels)
if routes:
try_config["routes"] = routes
try_task_config = {"version": 2, "parameters": params}
return try_task_config
def task_labels_from_try_config(try_task_config):
if try_task_config["version"] == 2:
parameters = try_task_config.get("parameters", {})
if "try_task_config" in parameters:
return parameters["try_task_config"].get("tasks")
else:
return None
elif try_task_config["version"] == 1:
return try_task_config.get("tasks", list())
else:
return None
def display_push_estimates(try_task_config):
task_labels = task_labels_from_try_config(try_task_config)
if task_labels is None:
return
cache_dir = os.path.join(
get_state_dir(specific_to_topsrcdir=True), "cache", "taskgraph"
)
graph_cache = None
dep_cache = None
target_file = None
for graph_cache_file in ["target_task_graph", "full_task_graph"]:
graph_cache = os.path.join(cache_dir, graph_cache_file)
if os.path.isfile(graph_cache):
dep_cache = graph_cache.replace("task_graph", "task_dependencies")
target_file = graph_cache.replace("task_graph", "task_set")
break
if not dep_cache:
return
download_task_history_data(cache_dir=cache_dir)
make_trimmed_taskgraph_cache(graph_cache, dep_cache, target_file=target_file)
durations = duration_summary(dep_cache, task_labels, cache_dir)
print(
"estimates: Runs {} tasks ({} selected, {} dependencies)".format(
durations["dependency_count"] + durations["selected_count"],
durations["selected_count"],
durations["dependency_count"],
)
)
print(
"estimates: Total task duration {}".format(
durations["dependency_duration"] + durations["selected_duration"]
)
)
if "percentile" in durations:
percentile = durations["percentile"]
if percentile > 50:
print("estimates: In the longest {}% of durations".format(100 - percentile))
else:
print("estimates: In the shortest {}% of durations".format(percentile))
print(
"estimates: Should take about {} (Finished around {})".format(
durations["wall_duration_seconds"],
durations["eta_datetime"].strftime("%Y-%m-%d %H:%M"),
)
)
# improves on `" ".join(sys.argv[:])` by requoting argv items containing spaces or single quotes
def get_sys_argv(injected_argv=None):
argv_to_use = injected_argv or sys.argv[:]
formatted_argv = []
for item in argv_to_use:
if " " in item or "'" in item:
formatted_item = f'"{item}"'
else:
formatted_item = item
formatted_argv.append(formatted_item)
return " ".join(formatted_argv)
def push_to_try(
method,
msg,
try_task_config=None,
stage_changes=False,
dry_run=False,
closed_tree=False,
files_to_change=None,
allow_log_capture=False,
push_to_lando=False,
):
push = not stage_changes and not dry_run
check_working_directory(push)
if try_task_config and method not in ("auto", "empty"):
try:
display_push_estimates(try_task_config)
except Exception:
traceback.print_exc()
print("warning: unable to display push estimates")
# Format the commit message
closed_tree_string = " ON A CLOSED TREE" if closed_tree else ""
the_cmdline = get_sys_argv()
full_commandline_entry = f"mach try command: `{the_cmdline}`"
commit_message = "{}{}\n\n{}\n\nPushed via `mach try {}`".format(
msg,
closed_tree_string,
full_commandline_entry,
method,
)
config_path = None
changed_files = []
if try_task_config:
if push and method not in ("again", "auto", "empty"):
write_task_config_history(msg, try_task_config)
config_path = write_task_config(try_task_config)
changed_files.append(config_path)
if (push or stage_changes) and files_to_change:
for path, content in files_to_change.items():
path = os.path.join(vcs.path, path)
with open(path, "wb") as fh:
fh.write(six.ensure_binary(content))
changed_files.append(path)
try:
if not push:
print("Commit message:")
print(commit_message)
if config_path:
print("Calculated try_task_config.json:")
with open(config_path) as fh:
print(fh.read())
return
vcs.add_remove_files(*changed_files)
try:
if push_to_lando:
push_to_lando_try(vcs, commit_message)
else:
vcs.push_to_try(commit_message, allow_log_capture=allow_log_capture)
except MissingVCSExtension as e:
if e.ext == "push-to-try":
print(HG_PUSH_TO_TRY_NOT_FOUND)
elif e.ext == "cinnabar":
print(GIT_CINNABAR_NOT_FOUND)
else:
raise
sys.exit(1)
finally:
if config_path and os.path.isfile(config_path):
os.remove(config_path)