| # -*- coding: utf-8 -*- |
| # Copyright 2019 The Fuchsia 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 recipe_engine import recipe_api |
| |
| from PB.go.chromium.org.luci.buildbucket.proto import ( |
| builds_service as builds_service_pb2, |
| ) |
| from PB.go.chromium.org.luci.buildbucket.proto import common as common_pb2 |
| |
| |
| class BuildbucketUtilApi(recipe_api.RecipeApi): |
| """Module for high-level buildbucket utilities specific to Fuchsia infra.""" |
| |
| # The maximum number of bytes allowed in a Buildbucket build's summary |
| # markdown. |
| MAX_SUMMARY_SIZE = 4000 |
| |
| @property |
| def id(self): |
| """A unique string identifier for the current build.""" |
| if self.m.led.launched_by_led: |
| return self.m.led.run_id |
| else: |
| return str(self.m.buildbucket.build.id) |
| |
| @property |
| def is_tryjob(self): |
| """Whether the current build is running as a tryjob.""" |
| bucket_name = self.m.buildbucket.build.builder.bucket |
| return bucket_name.endswith("try") |
| |
| @property |
| def build_url(self): |
| if self.m.led.launched_by_led: |
| return "https://ci.chromium.org/swarming/task/%s?server=%s" % ( |
| self.m.swarming.task_id, |
| self.m.buildbucket.build.infra.swarming.hostname, |
| ) |
| else: |
| return "https://ci.chromium.org/b/%d" % self.m.buildbucket.build.id |
| |
| def full_builder_name(self, builder=None): |
| builder = builder or self.m.buildbucket.build.builder |
| return "/".join((builder.project, builder.bucket, builder.builder)) |
| |
| def build_status(self, exc): |
| """Get corresponding build status for an exception.""" |
| if isinstance(exc, self.m.step.InfraFailure): |
| return common_pb2.INFRA_FAILURE |
| if isinstance(exc, self.m.step.StepFailure): |
| return common_pb2.FAILURE |
| return common_pb2.INFRA_FAILURE |
| |
| def last_build( |
| self, project=None, bucket=None, builder=None, fields=None, status=None |
| ): |
| """Returns the build proto for a builder's most recent successful build. |
| |
| Returns None if no successful builds could be found. |
| |
| Args: |
| project (str): LUCI project. If None, defaults to current build's |
| project. |
| bucket (str): BuildBucket bucket. If None, defaults to current build's |
| bucket. |
| builder (str): BuildBucket builder. If None, defaults to current |
| build's builder name. |
| fields (seq of str): BuildBucket Build proto message fields to include in |
| the result. If None, defaults to api.buildbucket.DEFAULT_FIELDS. |
| status (common_pb2.Status enum member): If set, will return the |
| last build with this status. |
| """ |
| predicate = builds_service_pb2.BuildPredicate() |
| predicate.builder.project = project or self.m.buildbucket.build.builder.project |
| predicate.builder.bucket = bucket or self.m.buildbucket.build.builder.bucket |
| predicate.builder.builder = builder or self.m.buildbucket.build.builder.builder |
| if status is not None: |
| predicate.status = status |
| |
| fields = fields or self.m.buildbucket.DEFAULT_FIELDS |
| builds = self.m.buildbucket.search(predicate, limit=1, fields=fields) |
| if not builds: |
| return None |
| |
| assert len(builds) == 1 |
| return builds[0] |
| |
| def summary_message( |
| self, |
| raw_text, |
| truncation_message, |
| max_line_length=200, |
| truncate_length=None, |
| escape_markdown=True, |
| ): |
| """Return a `summary_markdown`-friendly message from raw text. |
| |
| Args: |
| raw_text (str): Raw text to convert to summary message. |
| truncation_message (str): Message to attach to end of summary message |
| if the message was truncated. |
| max_line_length (int): Truncate lines longer than this value. |
| truncate_length (int): Maximum length of the summary message. |
| escape_markdown (bool): Whether to surround the summary in triple |
| backticks, causing it to render as raw monospace text in the |
| LUCI UI. |
| """ |
| if not truncate_length: |
| truncate_length = ( |
| self.MAX_SUMMARY_SIZE |
| # Include an extra buffer so that a parent build can expose its |
| # child's summary as its own summary, with a prefix. |
| - 150 |
| # Include an extra buffer for potential autocorrelator messages. |
| - 750 |
| ) |
| # Make space for the truncation message plus two newline characters. |
| truncate_length -= len(truncation_message) + 2 |
| # TODO(olivernewman): In Python 3 the length of a unicode character is |
| # always one regardless of how many bytes it contains, so we'll need to |
| # adjust max_line_length by `len(dots.encode())` to make it correct. |
| dots = "…" |
| truncated_line = False |
| failure_lines = [] |
| for line in raw_text.rstrip().split("\n"): |
| if len(line) > max_line_length: |
| line = unicode_truncate(line, max_line_length - len(dots)) + dots |
| truncated_line = True |
| failure_lines.append(line) |
| final_summary = "\n".join(failure_lines) |
| if len(final_summary) > truncate_length: |
| final_summary = ( |
| unicode_truncate(final_summary, truncate_length - len(dots)) + dots |
| ) |
| elif not truncated_line: |
| # No need to include a truncation warning if we didn't truncate |
| # any lines or the entire failure summary. |
| truncation_message = "" |
| |
| # Render the error details as code in the Milo build summary so |
| # that whitespace is respected; otherwise it would be ignored |
| # by Milo's markdown renderer. |
| if escape_markdown: |
| final_summary = "```\n%s\n```" % final_summary |
| |
| return final_summary + ( |
| "\n\n%s" % truncation_message if truncation_message else "" |
| ) |
| |
| |
| def unicode_truncate(string, max_bytes): |
| """Truncates a string while avoiding splitting unicode characters. |
| |
| The resulting string will have have len() <= max_bytes and will contain |
| valid UTF-8. |
| """ |
| chars = [] |
| byte_count = 0 |
| for unicode_char in string.decode("utf-8"): |
| encoded = unicode_char.encode("utf-8") |
| if byte_count + len(encoded) > max_bytes: |
| break |
| chars.append(encoded) |
| byte_count += len(encoded) |
| return "".join(chars) |
