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
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
import datetime
import functools
import logging
import mozpack.path as mozpath
from mozbuild.base import MozbuildObject
from taskgraph.optimize.base import OptimizationStrategy, register_strategy
from taskgraph.optimize.strategies import IndexSearch
from taskgraph.util.parameterization import resolve_timestamps
from gecko_taskgraph.optimize.mozlint import SkipUnlessMozlint
logger = logging.getLogger(__name__)
@register_strategy("skip-unless-schedules")
class SkipUnlessSchedules(OptimizationStrategy):
@functools.cache
def scheduled_by_push(self, files_changed):
mbo = MozbuildObject.from_environment()
# the decision task has a sparse checkout, so, mozbuild_reader will use
# a MercurialRevisionFinder with revision '.', which should be the same
# as `revision`; in other circumstances, it will use a default reader
rdr = mbo.mozbuild_reader(config_mode="empty")
components = set()
for p, m in rdr.files_info(files_changed).items():
components |= set(m["SCHEDULES"].components)
return components
def should_remove_task(self, task, params, conditions):
if params.get("pushlog_id") == -1:
return False
scheduled = self.scheduled_by_push(frozenset(params["files_changed"]))
conditions = set(conditions)
# if *any* of the condition components are scheduled, do not optimize
if conditions & scheduled:
return False
return True
@register_strategy("skip-unless-has-relevant-tests")
class SkipUnlessHasRelevantTests(OptimizationStrategy):
"""Optimizes tasks that don't run any tests that were
in child directories of a modified file.
"""
@functools.cache
def get_changed_dirs(self, files_changed):
changed = map(mozpath.dirname, files_changed)
# Filter out empty directories (from files modified in the root).
# Otherwise all tasks would be scheduled.
return {d for d in changed if d}
def should_remove_task(self, task, params, _):
if not task.attributes.get("test_manifests"):
return True
for d in self.get_changed_dirs(frozenset(params["files_changed"])):
for t in task.attributes["test_manifests"]:
if t.startswith(d):
logger.debug(
f"{task.label} runs a test path ({t}) contained by a modified file ({d})"
)
return False
return True
register_strategy("skip-unless-mozlint", args=("tools/lint",))(SkipUnlessMozlint)
@register_strategy("skip-unless-missing")
class SkipUnlessMissing(OptimizationStrategy):
"""Skips a task unless it is missing from a specified index.
This simply defers to Taskgraph's `index-search` optimization. The reason
we need this shim is because replacement and removal optimizations can't be
joined together in a composite strategy as removal and replacement happen
at different times.
"""
index_search = IndexSearch()
def _convert_datetime_str(self, dt):
if dt.endswith("Z"):
dt = dt[:-1]
return datetime.datetime.fromisoformat(dt).strftime(self.index_search.fmt)
def should_remove_task(self, task, params, index):
now = datetime.datetime.now(datetime.timezone.utc)
deadline = self._convert_datetime_str(
resolve_timestamps(now, task.task["deadline"])
)
return bool(
self.index_search.should_replace_task(task, params, deadline, [index])
)
@register_strategy("skip-constrained")
class SkipConstrained(OptimizationStrategy):
"""Skips tasks on constrained worker pools.
Always skips tasks that are on pools with capacity issues, typically
hardware.
"""
def should_remove_task(self, task, params, _):
if task.task["provisionerId"] == "releng-hardware":
return True
return False