# Copyright 2014 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import datetime
import functools
import hashlib
import logging
import os.path
from recipe_engine import config_types
from recipe_engine import recipe_api
from recipe_engine import util as recipe_util
import state
# TODO(borenet): This module was copied from build.git and heavily modified to
# remove dependencies on other modules in build.git. It belongs in a different
# repo. Remove this once it has been moved.
# Minimally supported version of swarming.py script (reported by --version).
MINIMAL_SWARMING_VERSION = (0, 8, 6)
def text_for_task(task):
lines = []
if task.dimensions.get('id'): # pragma: no cover
lines.append('Bot id: %r' % task.dimensions['id'])
if task.dimensions.get('os'):
lines.append('Run on OS: %r' % task.dimensions['os'])
return '<br/>'.join(lines)
def parse_time(value):
"""Converts serialized time from the API to datetime.datetime."""
# When microseconds are 0, the '.123456' suffix is elided. This means the
# serialized format is not consistent, which confuses the hell out of python.
# TODO(maruel): Remove third format once we enforce version >=0.8.2.
for fmt in ('%Y-%m-%dT%H:%M:%S.%f', '%Y-%m-%dT%H:%M:%S', '%Y-%m-%d %H:%M:%S'):
try:
return datetime.datetime.strptime(value, fmt)
except ValueError: # pragma: no cover
pass
raise ValueError('Failed to parse %s' % value) # pragma: no cover
class ReadOnlyDict(dict):
def __setitem__(self, key, value):
raise TypeError('ReadOnlyDict is immutable')
class SwarmingApi(recipe_api.RecipeApi):
"""Recipe module to use swarming.py tool to run tasks on Swarming.
General usage:
1. Tweak default task parameters applied to all swarming tasks (such as
default_dimensions and default_priority).
2. Isolate some test using 'isolate' recipe module. Get isolated hash as
a result of that process.
3. Create a task configuration using 'task(...)' method, providing
isolated hash obtained previously.
4. Tweak the task parameters. This step is optional.
5. Launch the task on swarming by calling 'trigger_task(...)'.
6. Continue doing useful work locally while the task is running concurrently
on swarming.
7. Wait for task to finish and collect its result (exit code, logs)
by calling 'collect_task(...)'.
See also example.py for concrete code.
"""
State = state.State
#############################################################################
# The below are helper functions to help transition between the old and new #
# swarming result formats. TODO(martiniss): remove these #
#############################################################################
def _is_expired(self, shard):
# FIXME: We really should only have one format for enums. We want to move to
# strings, currently have numbers.
return (
shard.get('state') == self.State.EXPIRED or
shard.get('state') == 'EXPIRED')
def _is_timed_out(self, shard):
# FIXME: We really should only have one format for enums. We want to move to
# strings, currently have numbers.
return (
shard.get('state') == self.State.TIMED_OUT or
shard.get('state') == 'TIMED_OUT')
def _get_exit_code(self, shard):
if shard.get('exit_code'):
return shard.get('exit_code') # pragma: no cover
lst = shard.get('exit_codes', [])
return str(lst[0]) if lst else None
def __init__(self, **kwargs):
super(SwarmingApi, self).__init__(**kwargs)
# All tests default to a x86-64 bot running with no GPU. This simplifies
# management so that new tests are not executed on exotic bots by accidents
# even if misconfigured.
self._default_dimensions = {
'cpu': 'x86-64',
'gpu': 'none',
}
# Expirations are set to mildly good values and will be tightened soon.
self._default_expiration = 60*60
self._default_env = {}
self._default_hard_timeout = 60*60
self._default_idempotent = False
self._default_io_timeout = 20*60
# The default priority is extremely low and should be increased dependending
# on the type of task.
self._default_priority = 200
self._default_tags = set()
self._default_user = None
self._pending_tasks = set()
self._show_isolated_out_in_collect_step = True
self._show_shards_in_collect_step = False
self._swarming_server = 'https://chromium-swarm.appspot.com'
self._verbose = False
@recipe_util.returns_placeholder
def summary(self):
return self.m.json.output()
@property
def swarming_server(self):
"""URL of Swarming server to use, default is a production one."""
return self._swarming_server
@swarming_server.setter
def swarming_server(self, value):
"""Changes URL of Swarming server to use."""
self._swarming_server = value
@property
def verbose(self):
"""True to run swarming scripts with verbose output."""
return self._verbose
@verbose.setter
def verbose(self, value):
"""Enables or disables verbose output in swarming scripts."""
assert isinstance(value, bool), value
self._verbose = value
@property
def default_expiration(self):
"""Number of seconds that the server will wait to find a bot able to run the
task.
If not bot runs the task by this number of seconds, the task is canceled as
EXPIRED.
This value can be changed per individual task.
"""
return self._default_expiration
@default_expiration.setter
def default_expiration(self, value):
assert 30 <= value <= 24*60*60, value
self._default_expiration = value
@property
def default_hard_timeout(self):
"""Number of seconds in which the task must complete.
If the task takes more than this amount of time, the process is assumed to
be hung. It forcibly killed via SIGTERM then SIGKILL after a grace period
(default: 30s). Then the task is marked as TIMED_OUT.
This value can be changed per individual task.
"""
return self._default_hard_timeout
@default_hard_timeout.setter
def default_hard_timeout(self, value):
assert 30 <= value <= 6*60*60, value
self._default_hard_timeout = value
@property
def default_io_timeout(self):
"""Number of seconds at which interval the task must write to stdout or
stderr.
If the task takes more than this amount of time between writes to stdout or
stderr, the process is assumed to be hung. It forcibly killed via SIGTERM
then SIGKILL after a grace period (default: 30s). Then the task is marked as
TIMED_OUT.
This value can be changed per individual task.
"""
return self._default_io_timeout
@default_io_timeout.setter
def default_io_timeout(self, value):
assert 30 <= value <= 6*60*60, value
self._default_io_timeout = value
@property
def default_idempotent(self):
"""Bool to specify if task deduplication can be done.
When set, the server will search for another task that ran in the last days
that had the exact same properties. If it finds one, the task will not be
run at all, the previous results will be returned as-is.
For more infos, see:
https://github.com/luci/luci-py/blob/master/appengine/swarming/doc/User-Guide.md#task-idempotency
This value can be changed per individual task.
"""
return self._default_idempotent
@default_idempotent.setter
def default_idempotent(self, value):
assert isinstance(value, bool), value
self._default_idempotent = value
@property
def default_user(self):
"""String to represent who triggered the task.
The user should be an email address when someone requested testing via
pre-commit or manual testing.
This value can be changed per individual task.
"""
return self._default_user
@default_user.setter
def default_user(self, value):
assert value is None or isinstance(value, basestring), value
self._default_user = value
@property
def default_dimensions(self):
"""Returns a copy of the default Swarming dimensions to run task on.
The dimensions are what is used to filter which bots are able to run the
task successfully. This is particularly useful to discern between OS
versions, type of CPU, GPU card or VM, or preallocated pool.
Example:
{'cpu': 'x86-64', 'os': 'Windows-XP-SP3'}
This value can be changed per individual task.
"""
return ReadOnlyDict(self._default_dimensions)
def set_default_dimension(self, key, value):
assert isinstance(key, basestring), key
assert isinstance(value, basestring) or value is None, value
if value is None:
self._default_dimensions.pop(key, None)
else:
self._default_dimensions[key] = value # pragma: no cover
@property
def default_env(self):
"""Returns a copy of the default environment variable to run tasks with.
By default the environment variable is not modified. Additional environment
variables can be specified for each task.
This value can be changed per individual task.
"""
return ReadOnlyDict(self._default_env)
def set_default_env(self, key, value):
assert isinstance(key, basestring), key
assert isinstance(value, basestring), value
self._default_env[key] = value
@property
def default_priority(self):
"""Swarming task priority for tasks triggered from the recipe.
Priority ranges from 1 to 255. The lower the value, the most important the
task is and will preempty any task with a lower priority.
This value can be changed per individual task.
"""
return self._default_priority
@default_priority.setter
def default_priority(self, value):
assert 1 <= value <= 255
self._default_priority = value
def add_default_tag(self, tag):
"""Adds a tag to the Swarming tasks triggered.
Tags are used for maintenance, they can be used to calculate the number of
tasks run for a day to calculate the cost of a type of type (CQ, ASAN, etc).
Tags can be added per individual task.
"""
assert ':' in tag, tag
self._default_tags.add(tag)
@property
def show_isolated_out_in_collect_step(self):
"""Show the shard's isolated out link in each collect step."""
return self._show_isolated_out_in_collect_step
@show_isolated_out_in_collect_step.setter
def show_isolated_out_in_collect_step(self, value):
self._show_isolated_out_in_collect_step = value
@property
def show_shards_in_collect_step(self):
"""Show the shard link in each collect step."""
return self._show_shards_in_collect_step
@show_shards_in_collect_step.setter
def show_shards_in_collect_step(self, value):
self._show_shards_in_collect_step = value
@staticmethod
def prefered_os_dimension(platform):
"""Given a platform name returns the prefered Swarming OS dimension.
Platform name is usually provided by 'platform' recipe module, it's one
of 'win', 'linux', 'mac'. This function returns more concrete Swarming OS
dimension that represent this platform on Swarming by default.
Recipes are free to use other OS dimension if there's a need for it. For
example WinXP try bot recipe may explicitly specify 'Windows-XP-SP3'
dimension.
"""
return {
'linux': 'Ubuntu-14.04',
'mac': 'Mac-10.9',
'win': 'Windows-7-SP1',
}[platform]
def task(self, title, isolated_hash, ignore_task_failure=False, shards=1,
task_output_dir=None, extra_args=None, idempotent=None,
cipd_packages=None, build_properties=None, merge=None):
"""Returns a new SwarmingTask instance to run an isolated executable on
Swarming.
For google test executables, use gtest_task() instead.
At the time of this writting, this code is used by V8, Skia and iOS.
The return value can be customized if necessary (see SwarmingTask class
below). Pass it to 'trigger_task' to launch it on swarming. Later pass the
same instance to 'collect_task' to wait for the task to finish and fetch its
results.
Args:
title: name of the test, used as part of a task ID.
isolated_hash: hash of isolated test on isolate server, the test should
be already isolated there, see 'isolate' recipe module.
ignore_task_failure: whether to ignore the test failure of swarming
tasks. By default, this is set to False.
shards: if defined, the number of shards to use for the task. By default
this value is either 1 or based on the title.
task_output_dir: if defined, the directory where task results are placed.
The caller is responsible for removing this folder when finished.
extra_args: list of command line arguments to pass to isolated tasks.
idempotent: whether this task is considered idempotent. Defaults
to self.default_idempotent if not specified.
cipd_packages: list of 3-tuples corresponding to CIPD packages needed for
the task: ('path', 'package_name', 'version'), defined as follows:
path: Path relative to the Swarming root dir in which to install
the package.
package_name: Name of the package to install,
eg. "infra/tools/authutil/${platform}"
version: Version of the package, either a package instance ID,
ref, or tag key/value pair.
build_properties: An optional dict containing various build properties.
These are typically but not necessarily the properties emitted by
bot_update.
merge: An optional dict containing:
"script": path to a script to call to post process and merge the
collected outputs from the tasks. The script should take one
named (but required) parameter, '-o' (for output), that represents
the path that the merged results should be written to, and accept
N additional paths to result files to merge. The merged results
should be in the JSON Results File Format
(https://www.chromium.org/developers/the-json-test-results-format)
and may optionally contain a top level "links" field that
may contain a dict mapping link text to URLs, for a set of
links that will be included in the buildbot output.
"args": an optional list of additional arguments to pass to the
above script.
"""
if idempotent is None:
idempotent = self.default_idempotent
return SwarmingTask(
title=title,
isolated_hash=isolated_hash,
dimensions=self._default_dimensions,
env=self._default_env,
priority=self.default_priority,
shards=shards,
buildername=self.m.properties.get('buildername'),
buildnumber=self.m.properties.get('buildnumber'),
user=self.default_user,
expiration=self.default_expiration,
io_timeout=self.default_io_timeout,
hard_timeout=self.default_hard_timeout,
idempotent=idempotent,
ignore_task_failure=ignore_task_failure,
extra_args=extra_args,
collect_step=self._default_collect_step,
task_output_dir=task_output_dir,
cipd_packages=cipd_packages,
build_properties=build_properties,
merge=merge)
def check_client_version(self, step_test_data=None):
"""Yields steps to verify compatibility with swarming_client version."""
return self.m.swarming_client.ensure_script_version(
'swarming.py', MINIMAL_SWARMING_VERSION, step_test_data)
def trigger_task(self, task, **kwargs):
"""Triggers one task.
It the task is sharded, will trigger all shards. This steps justs posts
the task and immediately returns. Use 'collect_task' to wait for a task to
finish and grab its result.
Behaves as a regular recipe step: returns StepData with step results
on success or raises StepFailure if step fails.
Args:
task: SwarmingTask instance.
kwargs: passed to recipe step constructor as-is.
"""
assert isinstance(task, SwarmingTask)
assert task.task_name not in self._pending_tasks, (
'Triggered same task twice: %s' % task.task_name)
assert 'os' in task.dimensions, task.dimensions
self._pending_tasks.add(task.task_name)
# Trigger parameters.
args = [
'trigger',
'--swarming', self.swarming_server,
'--isolate-server', self.m.isolate.isolate_server,
'--priority', str(task.priority),
'--shards', str(task.shards),
'--task-name', task.task_name,
'--dump-json', self.m.json.output(),
'--expiration', str(task.expiration),
'--io-timeout', str(task.io_timeout),
'--hard-timeout', str(task.hard_timeout),
]
for name, value in sorted(task.dimensions.iteritems()):
assert isinstance(value, basestring), value
args.extend(['--dimension', name, value])
for name, value in sorted(task.env.iteritems()):
assert isinstance(value, basestring), value
args.extend(['--env', name, value])
# Default tags.
tags = set(task.tags)
tags.update(self._default_tags)
tags.add('data:' + task.isolated_hash)
tags.add('name:' + task.title.split(' ')[0])
mastername = self.m.properties.get('mastername')
if mastername: # pragma: no cover
tags.add('master:' + mastername)
if task.buildername: # pragma: no cover
tags.add('buildername:' + task.buildername)
if task.buildnumber: # pragma: no cover
tags.add('buildnumber:%s' % task.buildnumber)
if task.dimensions.get('os'):
tags.add('os:' + task.dimensions['os'])
if self.m.properties.get('bot_id'): # pragma: no cover
tags.add('slavename:%s' % self.m.properties['bot_id'])
tags.add('stepname:%s' % self.get_step_name('', task))
rietveld = self.m.properties.get('rietveld')
issue = self.m.properties.get('issue')
patchset = self.m.properties.get('patchset')
if rietveld and issue and patchset:
# The expected format is strict to the usage of buildbot properties on the
# Chromium Try Server. Fix if necessary.
tags.add('rietveld:%s/%s/#ps%s' % (rietveld, issue, patchset))
for tag in sorted(tags):
assert ':' in tag, tag
args.extend(['--tag', tag])
if self.verbose:
args.append('--verbose')
if task.idempotent:
args.append('--idempotent')
if task.user:
args.extend(['--user', task.user])
if task.cipd_packages:
for path, pkg, version in task.cipd_packages:
args.extend(['--cipd-package', '%s:%s:%s' % (path, pkg, version)])
# What isolated command to trigger.
args.extend(('--isolated', task.isolated_hash))
# Additional command line args for isolated command.
if task.extra_args: # pragma: no cover
args.append('--')
args.extend(task.extra_args)
# The step can fail only on infra failures, so mark it as 'infra_step'.
try:
return self.m.python(
name=self.get_step_name('trigger', task),
script=self.m.swarming_client.path.join('swarming.py'),
args=args,
step_test_data=functools.partial(
self._gen_trigger_step_test_data, task),
infra_step=True,
**kwargs)
finally:
# Store trigger output with the |task|, print links to triggered shards.
step_result = self.m.step.active_result
step_result.presentation.step_text += text_for_task(task)
if step_result.presentation != self.m.step.FAILURE:
task._trigger_output = step_result.json.output
links = step_result.presentation.links
for index in xrange(task.shards):
url = task.get_shard_view_url(index)
if url:
links['shard #%d' % index] = url
assert not hasattr(step_result, 'swarming_task')
step_result.swarming_task = task
def collect_task(self, task, **kwargs):
"""Waits for a single triggered task to finish.
If the task is sharded, will wait for all shards to finish. Behaves as
a regular recipe step: returns StepData with step results on success or
raises StepFailure if task fails.
Args:
task: SwarmingTask instance, previously triggered with 'trigger' method.
kwargs: passed to recipe step constructor as-is.
"""
# TODO(vadimsh): Raise InfraFailure on Swarming failures.
assert isinstance(task, SwarmingTask)
assert task.task_name in self._pending_tasks, (
'Trying to collect a task that was not triggered: %s' %
task.task_name)
self._pending_tasks.remove(task.task_name)
try:
return task.collect_step(task, **kwargs)
finally:
try:
self.m.step.active_result.swarming_task = task
except Exception: # pragma: no cover
# If we don't have an active_result, something failed very early,
# so we eat this exception and let that one propagate.
pass
def trigger(self, tasks, **kwargs): # pragma: no cover
"""Batch version of 'trigger_task'.
Deprecated, to be removed soon. Use 'trigger_task' in a loop instead,
properly handling exceptions. This method doesn't handle trigger failures
well (it aborts on a first failure).
"""
return [self.trigger_task(t, **kwargs) for t in tasks]
def collect(self, tasks, **kwargs): # pragma: no cover
"""Batch version of 'collect_task'.
Deprecated, to be removed soon. Use 'collect_task' in a loop instead,
properly handling exceptions. This method doesn't handle collect failures
well (it aborts on a first failure).
"""
return [self.collect_task(t, **kwargs) for t in tasks]
# To keep compatibility with some build_internal code. To be removed as well.
collect_each = collect
@staticmethod
def _display_pending(summary_json, step_presentation):
"""Shows max pending time in seconds across all shards if it exceeds 10s."""
pending_times = [
(parse_time(shard['started_ts']) -
parse_time(shard['created_ts'])).total_seconds()
for shard in summary_json.get('shards', []) if shard.get('started_ts')
]
max_pending = max(pending_times) if pending_times else 0
# Only display annotation when pending more than 10 seconds to reduce noise.
if max_pending > 10:
step_presentation.step_text += '<br>swarming pending %ds' % max_pending
def _default_collect_step(
self, task, merged_test_output=None,
step_test_data=None,
**kwargs):
"""Produces a step that collects a result of an arbitrary task."""
task_output_dir = task.task_output_dir or self.m.raw_io.output_dir()
# If we don't already have a Placeholder, wrap the task_output_dir in one
# so we can read out of it later w/ step_result.raw_io.output_dir.
if not isinstance(task_output_dir, recipe_util.Placeholder):
task_output_dir = self.m.raw_io.output_dir(leak_to=task_output_dir)
task_args = [
'-o', merged_test_output or self.m.json.output(),
'--task-output-dir', task_output_dir,
]
merge_script = (task.merge.get('script')
or self.resource('noop_merge.py'))
merge_args = (task.merge.get('args') or [])
task_args.extend([
'--merge-script', merge_script,
'--merge-additional-args', self.m.json.dumps(merge_args),
])
if task.build_properties: # pragma: no cover
properties = dict(task.build_properties)
properties.update(self.m.properties)
task_args.extend([
'--build-properties', self.m.json.dumps(properties),
])
task_args.append('--')
# Arguments for the actual 'collect' command.
collect_cmd = [
'python',
'-u',
self.m.swarming_client.path.join('swarming.py'),
]
collect_cmd.extend(self.get_collect_cmd_args(task))
collect_cmd.extend([
'--task-summary-json', self.summary(),
])
task_args.extend(collect_cmd)
allowed_return_codes = {0}
if task.ignore_task_failure: # pragma: no cover
allowed_return_codes = 'any'
# The call to collect_task emits two JSON files:
# 1) a task summary JSON emitted by swarming
# 2) a gtest results JSON emitted by the task
# This builds an instance of StepTestData that covers both.
step_test_data = step_test_data or (
self.test_api.canned_summary_output(task.shards) +
self.m.json.test_api.output({}))
try:
with self.m.context(cwd=self.m.path['start_dir']):
return self.m.python(
name=self.get_step_name('', task),
script=self.resource('collect_task.py'),
args=task_args,
ok_ret=allowed_return_codes,
step_test_data=lambda: step_test_data,
**kwargs)
finally:
step_result = None
try:
step_result = self.m.step.active_result
step_result.presentation.step_text = text_for_task(task)
summary_json = step_result.swarming.summary
self._handle_summary_json(task, summary_json, step_result)
links = {}
if hasattr(step_result, 'json') and hasattr(step_result.json, 'output'):
links = step_result.json.output.get('links', {})
for k, v in links.iteritems(): # pragma: no cover
step_result.presentation.links[k] = v
except Exception as e:
if step_result:
step_result.presentation.logs['no_results_exc'] = [str(e)]
def get_step_name(self, prefix, task):
"""SwarmingTask -> name of a step of a waterfall.
Will take a task title (+ step name prefix) and append OS dimension to it.
Args:
prefix: prefix to append to task title, like 'trigger'.
task: SwarmingTask instance.
Returns:
'[<prefix>] <task title> on <OS>'
"""
prefix = '[%s] ' % prefix if prefix else ''
task_os = task.dimensions['os']
bot_os = self.prefered_os_dimension(self.m.platform.name)
suffix = ('' if (
task_os == bot_os or task_os.lower() == self.m.platform.name.lower())
else ' on %s' % task_os)
# Note: properly detecting dimensions of the bot the recipe is running
# on is somewhat non-trivial. It is not safe to assume it uses default
# or preferred dimensions for its OS. For example, the version of the OS
# can differ.
return ''.join((prefix, task.title, suffix))
def _handle_summary_json(self, task, summary, step_result):
# We store this now, and add links to all shards first, before failing the
# build. Format is tuple of (error message, shard that failed)
infra_failures = []
links = step_result.presentation.links
for index, shard in enumerate(summary['shards']):
url = task.get_shard_view_url(index)
display_text = 'shard #%d' % index
if not shard or shard.get('internal_failure'): # pragma: no cover
display_text = (
'shard #%d had an internal swarming failure' % index)
infra_failures.append((index, 'Internal swarming failure'))
elif self._is_expired(shard):
display_text = (
'shard #%d expired, not enough capacity' % index)
infra_failures.append((
index, 'There isn\'t enough capacity to run your test'))
elif self._is_timed_out(shard):
display_text = (
'shard #%d timed out, took too much time to complete' % index)
elif self._get_exit_code(shard) != '0': # pragma: no cover
display_text = 'shard #%d (failed)' % index
if self.show_isolated_out_in_collect_step:
isolated_out = shard.get('isolated_out')
if isolated_out:
link_name = 'shard #%d isolated out' % index
links[link_name] = isolated_out['view_url']
if url and self.show_shards_in_collect_step:
links[display_text] = url
self._display_pending(summary, step_result.presentation)
if infra_failures:
template = 'Shard #%s failed: %s'
# Done so that raising an InfraFailure doesn't cause an error.
# TODO(martiniss): Remove this hack. Requires recipe engine change
step_result._retcode = 2
step_result.presentation.status = self.m.step.EXCEPTION
raise recipe_api.InfraFailure(
'\n'.join(template % f for f in infra_failures), result=step_result)
def get_collect_cmd_args(self, task):
"""SwarmingTask -> argument list for 'swarming.py' command."""
args = [
'collect',
'--swarming', self.swarming_server,
'--decorate',
'--print-status-updates',
]
if self.verbose:
args.append('--verbose')
args.extend(('--json', self.m.json.input(task.trigger_output)))
return args
def _gen_trigger_step_test_data(self, task):
"""Generates an expected value of --dump-json in 'trigger' step.
Used when running recipes to generate test expectations.
"""
# Suffixes of shard subtask names.
subtasks = []
if task.shards == 1:
subtasks = ['']
else:
subtasks = [':%d:%d' % (task.shards, i) for i in range(task.shards)]
return self.m.json.test_api.output({
'base_task_name': task.task_name,
'tasks': {
'%s%s' % (task.task_name, suffix): {
'task_id': '1%02d00' % i,
'shard_index': i,
'view_url': '%s/user/task/1%02d00' % (self.swarming_server, i),
} for i, suffix in enumerate(subtasks)
},
})
class SwarmingTask(object):
"""Definition of a task to run on swarming."""
def __init__(self, title, isolated_hash, ignore_task_failure, dimensions,
env, priority, shards, buildername, buildnumber, expiration,
user, io_timeout, hard_timeout, idempotent, extra_args,
collect_step, task_output_dir, cipd_packages=None,
build_properties=None, merge=None):
"""Configuration of a swarming task.
Args:
title: display name of the task, hints to what task is doing. Usually
corresponds to a name of a test executable. Doesn't have to be unique.
isolated_hash: hash of isolated file that describes all files needed to
run the task as well as command line to launch. See 'isolate' recipe
module.
ignore_task_failure: whether to ignore the test failure of swarming
tasks.
cipd_packages: list of 3-tuples corresponding to CIPD packages needed for
the task: ('path', 'package_name', 'version'), defined as follows:
path: Path relative to the Swarming root dir in which to install
the package.
package_name: Name of the package to install,
eg. "infra/tools/authutil/${platform}"
version: Version of the package, either a package instance ID,
ref, or tag key/value pair.
collect_step: callback that will be called to collect and processes
results of task execution, signature is collect_step(task, **kwargs).
dimensions: key-value mapping with swarming dimensions that specify
on what Swarming slaves task can run. One important dimension is 'os',
which defines platform flavor to run the task on. See Swarming doc.
env: key-value mapping with additional environment variables to add to
environment before launching the task executable.
priority: integer [0, 255] that defines how urgent the task is.
Lower value corresponds to higher priority. Swarming service executes
tasks with higher priority first.
shards: how many concurrent shards to run, makes sense only for
isolated tests based on gtest. Swarming uses GTEST_SHARD_INDEX
and GTEST_TOTAL_SHARDS environment variables to tell the executable
what shard to run.
buildername: buildbot builder this task was triggered from.
buildnumber: build number of a build this task was triggered from.
expiration: number of schedule until the task shouldn't even be run if it
hadn't started yet.
user: user that requested this task, if applicable.
io_timeout: number of seconds that the task is allowed to not emit any
stdout bytes, after which it is forcibly killed.
hard_timeout: number of seconds for which the task is allowed to run,
after which it is forcibly killed.
idempotent: True if the results from a previous task can be reused. E.g.
this task has no side-effects.
extra_args: list of command line arguments to pass to isolated tasks.
task_output_dir: if defined, the directory where task results are placed
during the collect step.
build_properties: An optional dict containing various build properties.
These are typically but not necessarily the properties emitted by
bot_update.
merge: An optional dict containing:
"script": path to a script to call to post process and merge the
collected outputs from the tasks.
"args": an optional list of additional arguments to pass to the
above script.
"""
self._trigger_output = None
self.build_properties = build_properties
self.buildername = buildername
self.buildnumber = buildnumber
self.cipd_packages = cipd_packages
self.collect_step = collect_step
self.dimensions = dimensions.copy()
self.env = env.copy()
self.expiration = expiration
self.extra_args = tuple(extra_args or [])
self.hard_timeout = hard_timeout
self.idempotent = idempotent
self.ignore_task_failure = ignore_task_failure
self.io_timeout = io_timeout
self.isolated_hash = isolated_hash
self.merge = merge or {}
self.priority = priority
self.shards = shards
self.tags = set()
self.task_output_dir = task_output_dir
self.title = title
self.user = user
@property
def task_name(self):
"""Name of this task, derived from its other properties.
The task name is purely to make sense of the task and is not used in any
other way.
"""
out = '%s/%s/%s' % (
self.title, self.dimensions['os'], self.isolated_hash[:10])
if self.buildername: # pragma: no cover
out += '/%s/%s' % (self.buildername, self.buildnumber or -1)
return out
@property
def trigger_output(self):
"""JSON results of 'trigger' step or None if not triggered."""
return self._trigger_output
def get_shard_view_url(self, index):
"""Returns URL of HTML page with shard details or None if not available.
Works only after the task has been successfully triggered.
"""
if self._trigger_output and self._trigger_output.get('tasks'):
for shard_dict in self._trigger_output['tasks'].itervalues():
if shard_dict['shard_index'] == index:
return shard_dict['view_url']