blob: ae8cdb6f481499534ff7938a4081bccfcfb3807f [file] [log] [blame]
# Copyright 2020 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.
from google.protobuf import json_format
from past.builtins import basestring
from recipe_engine import recipe_api
from import build as build_pb2
class Job(object):
"""Describes a job that could be launched via led.
name (str): Name of the job.
properties (dict): Properties as the arguments to `led edit -p`.
dimensions (dict): Dimensions as the arguments to `led edit -d`.
task_id (str): Id of the swarming task.
task_server (str): Host name of the swarming server, e.g.
task_result (api.swarming.TaskResult): Result of the swarming task.
build_proto(build_pb2.Build): Proto generated from a buildbucket build.
outcome (str): The outcome could be "success", "none" when task_result.state
is None, or TaskState in lower case.
def __init__(self, name):
# Metadata attached before execution.
assert isinstance(name, basestring) = name = {'name': name}
self.dimensions = {}
# Metadata attached during execution.
self.task_id = None
self.task_server = None
# Metadata attached after execution.
self.task_result = None
self.build_proto = None
self.outcome = None
def task_url(self):
"""Returns the URL of the associated task in the Swarming UI."""
return "%s/task?id=%s" % (self.task_server, self.task_id)
def milo_url(self):
"""Returns the URL of the associated task in the Milo UI."""
return "" % (
self.task_id, self.task_server)
class JobApi(recipe_api.RecipeApi):
"""API for launching jobs and collecting the results."""
def __init__(self, *args, **kwargs):
super(JobApi, self).__init__(*args, **kwargs)
def new(self, job_name):
return Job(job_name)
def current_recipe(self):
def launch(self, job, presentation):
"""Launches a job with led.
job (Job): The job definition.
presentation (StepPresentation): The presentation to add logs to.
The input job object with additional details about the execution.
current =
led_data = self.m.led(
"luci.%s.%s:%s" % (current.project, current.bucket, current.builder),
edit_args = []
for k, v in sorted(
edit_args.extend(["-p", "%s=%s" % (k, self.m.json.dumps(v))])
for k, v in sorted(job.dimensions.items()):
edit_args.extend(["-d", "%s=%s" % (k, v)])
led_data = led_data.then("edit", *edit_args)
led_data = self.m.led.inject_input_recipes(led_data)
final = led_data.then("launch", "-modernize")
job.task_id = final.launch_result.task_id
job.task_server = final.launch_result.swarming_hostname
presentation.links[] = job.milo_url
return job
def collect(self, jobs, presentation):
"""Collects execution metadata for a list of jobs.
jobs (list(Job)): The jobs to collect information for.
presentation (StepPresentation): The presentation to add logs to.
The input jobs with additional details collected from execution.
by_task_id = {job.task_id: job for job in jobs}
swarming_results = self.m.swarming.collect(
for result in sorted(swarming_results, key=lambda x:
job = by_task_id[]
job.task_result = result
# Led launch ensures this file is present in the task root dir.
build_proto_path = result.output_dir.join("build.proto.json")
build_proto = self.m.file.read_proto(
"read build.proto.json", build_proto_path, build_pb2.Build, "JSONPB")
job.build_proto = build_proto
if result.success:
job.outcome = "success"
elif not result.state:
job.outcome = "none"
# Example result.state: TaskState.COMPLETED
job.outcome = str(result.state)[10:].lower()
presentation.links["%s (%s)" % (, job.outcome)] = job.milo_url
return jobs