gcpdiag.queries.cloudrun

29class Service(models.Resource):
30  """Represents Cloud Run service."""
31  _resource_data: dict
32
33  def __init__(self, project_id, resource_data):
34    super().__init__(project_id=project_id)
35    self._resource_data = resource_data
36
37  @property
38  def name(self) -> str:
39    m = re.search(r'/services/([^/]+)$', self._resource_data['name'])
40    if not m:
41      raise RuntimeError('can\'t determine name of service %s' %
42                         (self._resource_data['name']))
43    return m.group(1)
44
45  @property
46  def id(self) -> str:
47    return self._resource_data['uid']
48
49  @property
50  def full_path(self) -> str:
51    return self._resource_data['name']
52
53  @property
54  def conditions(self) -> Dict[str, 'ServiceCondition']:
55    return {
56        condition['type']: ServiceCondition(condition)
57        for condition in self._resource_data['conditions']
58    }
59
60  @property
61  def short_path(self) -> str:
62    path = self.project_id + '/' + self.id
63    return path

66class ServiceCondition:
67  """Represents Cloud Run service status condition."""
68  _resource_data: dict
69
70  def __init__(self, resource_data):
71    self._resource_data = resource_data
72
73  @property
74  def message(self) -> str:
75    return self._resource_data.get('message', '')

 78def get_all_locations(project_id: str) -> Iterable[str]:
 79  """Return list of all regions
 80
 81  Args:
 82      project_id (str): project id for this request
 83
 84  Raises:
 85      utils.GcpApiError: Raises GcpApiError in case of query issues
 86
 87  Returns:
 88      Iterable[Region]: Return list of all regions
 89  """
 90  try:
 91    cloudrun_api = apis.get_api('run', 'v1', project_id)
 92    request = cloudrun_api.projects().locations().list(
 93        name=f'projects/{project_id}')
 94    response = request.execute(num_retries=config.API_RETRIES)
 95    if not response or 'locations' not in response:
 96      return set()
 97
 98    return {
 99        location['name']
100        for location in response['locations']
101        if 'name' in location
102    }
103  except googleapiclient.errors.HttpError as err:
104    raise utils.GcpApiError(err) from err

107@caching.cached_api_call
108def get_services(context: models.Context) -> Mapping[str, Service]:
109  """Get a list of Cloud Run services matching the given context,
110  indexed by service id."""
111  services: Dict[str, Service] = {}
112
113  if not apis.is_enabled(context.project_id, 'run'):
114    return services
115
116  locations = get_all_locations(context.project_id)
117
118  for location in locations:
119    m = re.search(r'/locations/([^/]+)$', location)
120    if not m:
121      continue
122    region = m.group(1)
123    cloudrun_api = apis.get_api('run', 'v2', context.project_id)
124    logging.info(
125        'fetching list of cloud run services in the project %s for the region %s',
126        context.project_id, region)
127    query = cloudrun_api.projects().locations().services().list(
128        parent=f'projects/{context.project_id}/locations/{region}')
129    try:
130      resp = query.execute(num_retries=config.API_RETRIES)
131      if 'services' not in resp:
132        continue
133      for s in resp['services']:
134        # projects/{project}/locations/{location}/services/{serviceId}.
135        result = re.match(r'projects/[^/]+/locations/([^/]+)/services/([^/]+)',
136                          s['name'])
137        if not result:
138          logging.error('invalid cloudrun name: %s', s['name'])
139          raise RuntimeError(
140              'missing data in projects.locations.services.list response')
141        location = result.group(1)
142        labels = s.get('labels', {})
143        name = result.group(2)
144        if not context.match_project_resource(
145            location=location, labels=labels, resource=name):
146          continue
147
148        services[s['uid']] = Service(project_id=context.project_id,
149                                     resource_data=s)
150    except googleapiclient.errors.HttpError as err:
151      raise utils.GcpApiError(err) from err
152  return services

155def get_service(project_id: str, region: str, service_name: str) -> Service:
156  cloudrun_api = apis.get_api('run', 'v2', project_id)
157  request = cloudrun_api.projects().locations().services().get(
158      name=f'projects/{project_id}/locations/{region}/services/{service_name}')
159  response = request.execute(num_retries=config.API_RETRIES)
160  return Service(project_id, response)