gcpdiag.queries.gcb

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

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

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

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

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

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