gcpdiag.queries.gcb

62@dataclasses.dataclass(frozen=True)
63class BuildOptions:
64  """representation of build.options object"""
65
66  logging: str
67  log_streaming_option: str
68
69  def is_bucket_streaming_enabled(self) -> bool:
70    return self.logging != 'GCS_ONLY' or self.log_streaming_option != 'STREAM_OFF'

73class BuildOptionsBuilder:
74  """Build options builder from dictionary."""
75
76  def __init__(self, options: dict):
77    self._options = options
78
79  def build(self) -> BuildOptions:
80    return BuildOptions(
81      logging=self._get_logging(),
82      log_streaming_option=self._get_log_streaming_option(),
83    )
84
85  def _get_logging(self) -> str:
86    return self._options.get('logging', 'LEGACY')
87
88  def _get_log_streaming_option(self) -> str:
89    return self._options.get('logStreamingOption', 'LOGGING_UNSPECIFIED')

92@dataclasses.dataclass(frozen=True)
93class FailureInfo:
94  """Wrapper around build.failureInfo object."""
95
96  failure_type: str

 99class FailureInfoBuilder:
100  """Wrapper around build.failureInfo object."""
101
102  def __init__(self, failure_info: dict):
103    self._failure_info = failure_info
104
105  def build(self) -> FailureInfo:
106    return FailureInfo(failure_type=self._get_failure_type())
107
108  def _get_failure_type(self) -> str:
109    return self._failure_info.get('type', '')

112class Build(models.Resource):
113  """Represents a Cloud Build execution."""
114
115  _resource_data: dict
116
117  def __init__(self, project_id, location, resource_data):
118    super().__init__(project_id=project_id)
119    self.location = location
120    self._resource_data = resource_data
121
122  @property
123  def id(self) -> str:
124    return self._resource_data['id']
125
126  @property
127  def full_path(self) -> str:
128    return f'projects/{self.project_id}/locations/{self.location}/builds/{self.id}'
129
130  @property
131  def short_path(self) -> str:
132    path = self.project_id + '/' + self.id
133    return path
134
135  @property
136  def status(self) -> str:
137    return self._resource_data['status']
138
139  @property
140  def service_account(self) -> Optional[str]:
141    return self._resource_data.get('serviceAccount')
142
143  @property
144  def images(self) -> List[str]:
145    return self._resource_data.get('images', [])
146
147  @property
148  def logs_bucket(self) -> str:
149    return self._resource_data.get('logsBucket', '')
150
151  @property
152  def options(self) -> BuildOptions:
153    return BuildOptionsBuilder(self._resource_data.get('options', {})).build()
154
155  @property
156  def failure_info(self) -> FailureInfo:
157    return FailureInfoBuilder(self._resource_data.get('failureInfo', {})).build()

160class Trigger(models.Resource):
161  """Represents a Cloud Build trigger instance."""
162
163  _resource_data: dict
164
165  def __init__(self, project_id, resource_data):
166    super().__init__(project_id=project_id)
167    self._resource_data = resource_data
168
169  @property
170  def name(self) -> str:
171    if 'name' not in self._resource_data:
172      return ''
173    return self._resource_data['name']
174
175  @property
176  def id(self) -> str:
177    return self._resource_data['id']
178
179  @property
180  def full_path(self) -> str:
181    return f'projects/{self.project_id}/locations/-/triggers/{self.id}'
182
183  @property
184  def short_path(self) -> str:
185    path = self.project_id + '/' + self.id
186    return path

189@caching.cached_api_call
190def get_builds(context: models.Context) -> Mapping[str, Build]:
191  """Get a list of Cloud Build instances matching the given context, indexed by Cloud Build id."""
192  if not apis.is_enabled(context.project_id, 'cloudbuild'):
193    return {}
194  build_api = apis.get_api('cloudbuild', 'v1', context.project_id)
195  batch = []
196  builds = {}
197  start_time = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(
198    days=config.get('within_days')
199  )
200  logging.debug('fetching list of builds in the project %s', context.project_id)
201  for location in LOCATIONS:
202    query = (
203      build_api.projects()
204      .locations()
205      .builds()
206      .list(
207        parent=f'projects/{context.project_id}/locations/{location}',
208        filter=f'create_time>"{start_time.isoformat()}"',
209      )
210    )
211    batch.append(query)
212  for request, response, exception in apis_utils.execute_concurrently(
213    api=build_api, requests=batch, context=context
214  ):
215    if exception:
216      if isinstance(exception, googleapiclient.errors.HttpError):
217        raise utils.GcpApiError(exception) from exception
218      else:
219        raise exception
220    if request is None or not hasattr(request, 'uri'):
221      logging.warning('Skipping request in batch, invalid request: %s', request)
222      continue
223    match = re.search(r'projects/([^/]+)/locations/([^/]+)', request.uri)
224    assert match, 'Bug: request uri does not match respected format'
225    project_id = match.group(1)
226    location = match.group(2)
227    if response is None or 'builds' not in response:
228      continue
229    for build in response['builds']:
230      # verify that we have some minimal data that we expect
231      if 'id' not in build:
232        raise RuntimeError('missing data in projects.locations.builds.list response')
233      r = re.search(r'projects/([^/]+)/locations/([^/]+)/builds/([^/]+)', build['name'])
234      if not r:
235        logging.error('build has invalid data: %s', build['name'])
236        continue
237
238      if not context.match_project_resource(resource=r.group(3)):
239        continue
240
241      builds[build['id']] = Build(project_id=project_id, location=location, resource_data=build)
242  return builds

245@caching.cached_api_call
246def get_triggers(context: models.Context) -> Mapping[str, Trigger]:
247  """Get a list of Cloud Build triggers matching the given context,
248  indexed by Cloud Build trigger id."""
249  triggers: Dict[str, Trigger] = {}
250  if not apis.is_enabled(context.project_id, 'cloudbuild'):
251    return triggers
252  build_api = apis.get_api('cloudbuild', 'v1', context.project_id)
253  logging.debug('fetching list of triggers in the project %s', context.project_id)
254  query = (
255    build_api.projects()
256    .locations()
257    .triggers()
258    .list(parent=f'projects/{context.project_id}/locations/global')
259  )
260  try:
261    resp = query.execute(num_retries=config.API_RETRIES)
262    if 'triggers' not in resp:
263      return triggers
264    for resp_f in resp['triggers']:
265      # verify that we have some minimal data that we expect
266      if 'id' not in resp_f:
267        raise RuntimeError('missing data in projects.locations.triggers.list response')
268      f = Trigger(project_id=context.project_id, resource_data=resp_f)
269      triggers[f.id] = f
270  except googleapiclient.errors.HttpError as err:
271    raise utils.GcpApiError(err) from err
272  return triggers