blob: d30f46560eb56ad89450ff2b0b6eb7c2ae681c2e [file] [log] [blame]
# Copyright 2018 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 json
import functools
from google.appengine.ext import ndb
from google.protobuf import json_format
from google.protobuf import symbol_database
from components import auth
from components import protoutil
from components import prpc
from components import utils
# Some of these imports are required to populate proto symbol db.
from proto import common_pb2
from proto import build_pb2 # pylint: disable=unused-import
from proto import rpc_pb2 # pylint: disable=unused-import
from proto import rpc_prpc_pb2
from proto import step_pb2 # pylint: disable=unused-import
from v2 import tokens
from v2 import validation
from v2 import default_field_masks
import buildtags
import config
import model
import search
import service
import user
import v2
# Header for passing token to authenticate build messages, e.g. UpdateBuild RPC.
# Lowercase because metadata is stored in lowercase.
BUILD_TOKEN_HEADER = 'x-build-token'
class StatusError(Exception):
def __init__(self, code, *details_and_args):
if details_and_args:
details = details_and_args[0] % details_and_args[1:]
details = code[1]
super(StatusError, self).__init__('%s: %s' % (code, details))
self.code = code
self.details = details
not_found = lambda *args: StatusError(prpc.StatusCode.NOT_FOUND, *args)
invalid_argument = (
lambda *args: StatusError(prpc.StatusCode.INVALID_ARGUMENT, *args)
for m in rpc_prpc_pb2.BuildsServiceDescription['service_descriptor'].method
def rpc_impl_async(rpc_name):
"""Returns a decorator for an async Builds RPC implementation.
Handles auth.AuthorizationError and StatusError.
Adds fourth method argument to the method, a protoutil.Mask.
If request has "fields" field, treats it as a FieldMask, parses it to a
protoutil.Mask and passes that.
After the method returns a response, the response is trimmed according to the
mask. Requires request message to have "fields" field of type FieldMask.
The default field masks are defined in default_field_masks.MASKS.
method_desc = METHODS_BY_NAME[rpc_name]
res_class = symbol_database.Default().GetSymbol(method_desc.output_type[1:])
default_mask = default_field_masks.MASKS.get(res_class)
def decorator(fn_async):
def decorated(req, ctx):
mask = default_mask
# Require that all RPC requests have "fields" field mask.
if req.HasField('fields'):
mask = protoutil.Mask.from_field_mask(
req.fields, res_class.DESCRIPTOR
except ValueError as ex:
raise invalid_argument('invalid fields: %s', ex)
res = yield fn_async(req, ctx, mask)
if mask: # pragma: no branch
raise ndb.Return(res)
except auth.AuthorizationError:
raise not_found()
except validation.Error as ex:
raise invalid_argument('%s', ex.message)
except StatusError as ex:
raise ndb.Return(None)
return decorated
return decorator
def bucket_id_string(builder_id):
return config.format_bucket_id(builder_id.project, builder_id.bucket)
def builds_to_v2_async(builds, build_mask):
"""Converts model.Build instances to build_pb2.Build messages."""
builds_msgs = map(v2.build_to_v2, builds)
if build_mask and build_mask.includes('steps'): # pragma: no branch
build_steps_list = yield ndb.get_multi_async([
model.BuildSteps.key_for(b.key) for b in builds
for b, build_steps in zip(builds_msgs, build_steps_list):
if build_steps: # pragma: no branch
raise ndb.Return(builds_msgs)
def build_predicate_to_search_query(predicate):
"""Converts a rpc_pb2.BuildPredicate to search.Query.
Assumes predicate is valid.
q = search.Query(
tags=[buildtags.unparse(p.key, p.value) for p in predicate.tags],
created_by=predicate.created_by or None,
if predicate.status == common_pb2.ENDED_MASK else predicate.status
# Filter by builder.
if predicate.HasField('builder'):
if predicate.builder.bucket:
q.bucket_ids = [bucket_id_string(predicate.builder)]
q.project = predicate.builder.project
if predicate.builder.builder:
buildtags.unparse(buildtags.BUILDER_KEY, predicate.builder.builder)
# Filter by gerrit changes.
buildsets = [
buildtags.gerrit_change_buildset(, c.change, c.patchset)
for c in predicate.gerrit_changes
q.tags.extend(buildtags.unparse(buildtags.BUILDSET_KEY, b) for b in buildsets)
# Filter by creation time.
if predicate.create_time.HasField('start_time'):
q.create_time_low = predicate.create_time.start_time.ToDatetime()
if predicate.create_time.HasField('end_time'):
q.create_time_high = predicate.create_time.end_time.ToDatetime()
# Filter by build range.
if predicate.HasField('build'):
# 0 means no boundary.
q.build_low = or None
q.build_high = or None
return q
def get_build_async(req, _ctx, mask):
"""Retrieves a build by id or number."""
build_v1 = yield service.get_async(
tag = buildtags.build_address_tag(
# TODO( migrate build_address to use short buckets.
'luci.%s.%s' % (req.builder.project, req.builder.bucket),
found, _ = yield search.search_async(
search.Query(bucket_ids=[bucket_id_string(req.builder)], tags=[tag])
build_v1 = found[0] if found else None
if not build_v1:
raise not_found()
raise ndb.Return((yield builds_to_v2_async([build_v1], mask))[0])
def search_builds_async(req, _ctx, mask):
"""Searches for builds."""
q = build_predicate_to_search_query(req.predicate)
q.max_builds = req.page_size or None
q.start_cursor = req.page_token
builds_v1, cursor = yield search.search_async(q)
raise ndb.Return(
yield builds_to_v2_async(builds_v1, mask.submask('builds.*'))
def validate_build_token(req, ctx):
"""Validates build token stored in RPC metadata."""
metadata = dict(ctx.invocation_metadata())
token = metadata.get(BUILD_TOKEN_HEADER)
if not token:
raise StatusError(
'missing token in build update request',
except auth.InvalidTokenError as e:
raise StatusError(prpc.StatusCode.UNAUTHENTICATED, '%s', e.message)
def update_build_async(req, ctx, _mask):
"""Update build as in given request.
For now, only update build steps.
validate_build_token(req, ctx)
if not (yield user.can_update_build_async()):
raise StatusError(
prpc.StatusCode.PERMISSION_DENIED, 'user not permitted to update build'
update_paths = set(req.update_mask.paths)
def txn_async():
build_proto =
# Get existing build.
build = yield service.get_async(
if not build:
raise not_found(
'Cannot update nonexisting build with id %s',
to_put = [build]
if 'build.steps' not in update_paths:
build_steps = yield model.BuildSteps.key_for(build.key).get_async()
# Update build steps.
build_steps = model.BuildSteps(
if '' in update_paths:
# TODO(nodir): persist it in a separate entity, in Struct binary format.
# The following is inefficient.
build.result_details = build.result_details or {}
build.result_details[model.PROPERTIES_PARAMETER] = json.loads(
# Store and convert back to build_pb2.Build proto for return.
yield ndb.put_multi_async(to_put)
raise ndb.Return(v2.build_to_v2(build, build_steps))
build = yield txn_async()
raise ndb.Return(build)
# Maps an rpc_pb2.BatchRequest.Request field name to an async function
# (req, ctx) => ndb.Future of res.
'get_build': get_build_async,
'search_builds': search_builds_async,
assert set(BATCH_REQUEST_TYPE_TO_RPC_IMPL) == set(
class BuildsApi(object):
"""Implements buildbucket.v2.Builds proto service."""
# "mask" parameter in RPC implementations is added by rpc_impl_async.
# pylint: disable=no-value-for-parameter
DESCRIPTION = rpc_prpc_pb2.BuildsServiceDescription
def GetBuild(self, req, ctx):
return get_build_async(req, ctx).get_result()
def SearchBuilds(self, req, ctx):
return search_builds_async(req, ctx).get_result()
def UpdateBuild(self, req, ctx):
return update_build_async(req, ctx).get_result()
def Batch(self, req, ctx):
def serve_subrequest_async(sub_req):
request_type = sub_req.WhichOneof('request')
sub_res = rpc_pb2.BatchResponse.Response()
if not request_type:
sub_res.error.code = prpc.StatusCode.INVALID_ARGUMENT.value
sub_res.error.message = 'request is not specified'
raise ndb.Return(sub_res)
rpc_impl = BATCH_REQUEST_TYPE_TO_RPC_IMPL[request_type]
sub_ctx = ctx.clone()
rpc_res = yield rpc_impl(getattr(sub_req, request_type), sub_ctx)
if sub_ctx.code != prpc.StatusCode.OK:
sub_res.error.code = sub_ctx.code.value
sub_res.error.message = sub_ctx.details
getattr(sub_res, request_type).MergeFrom(rpc_res)
raise ndb.Return(sub_res)
return rpc_pb2.BatchResponse(
for _, r in utils.async_apply(req.requests, serve_subrequest_async)