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 short_path(self) -> str:
55    path = self.project_id + '/' + self.id
56    return path

59def get_all_locations(project_id: str) -> Iterable[str]:
60  """Return list of all regions
61
62  Args:
63      project_id (str): project id for this request
64
65  Raises:
66      utils.GcpApiError: Raises GcpApiError in case of query issues
67
68  Returns:
69      Iterable[Region]: Return list of all regions
70  """
71  try:
72    cloudrun_api = apis.get_api('run', 'v1', project_id)
73    request = cloudrun_api.projects().locations().list(
74        name=f'projects/{project_id}')
75    response = request.execute(num_retries=config.API_RETRIES)
76    if not response or 'locations' not in response:
77      return set()
78
79    return {
80        location['name']
81        for location in response['locations']
82        if 'name' in location
83    }
84  except googleapiclient.errors.HttpError as err:
85    raise utils.GcpApiError(err) from err

 88@caching.cached_api_call
 89def get_services(context: models.Context) -> Mapping[str, Service]:
 90  """Get a list of Cloud Run services matching the given context,
 91  indexed by service id."""
 92  services: Dict[str, Service] = {}
 93
 94  if not apis.is_enabled(context.project_id, 'run'):
 95    return services
 96
 97  locations = get_all_locations(context.project_id)
 98
 99  for location in locations:
100    m = re.search(r'/locations/([^/]+)$', location)
101    if not m:
102      continue
103    region = m.group(1)
104    cloudrun_api = apis.get_api('run', 'v2', context.project_id)
105    logging.info(
106        'fetching list of cloud run services in the project %s for the region %s',
107        context.project_id, region)
108    query = cloudrun_api.projects().locations().services().list(
109        parent=f'projects/{context.project_id}/locations/{region}')
110    try:
111      resp = query.execute(num_retries=config.API_RETRIES)
112      if 'services' not in resp:
113        continue
114      for s in resp['services']:
115        # projects/{project}/locations/{location}/services/{serviceId}.
116        result = re.match(r'projects/[^/]+/locations/([^/]+)/services/([^/]+)',
117                          s['name'])
118        if not result:
119          logging.error('invalid cloudrun name: %s', s['name'])
120          raise RuntimeError(
121              'missing data in projects.locations.services.list response')
122        location = result.group(1)
123        labels = s.get('labels', {})
124        name = result.group(2)
125        if not context.match_project_resource(
126            location=location, labels=labels, resource=name):
127          continue
128
129        services[s['uid']] = Service(project_id=context.project_id,
130                                     resource_data=s)
131    except googleapiclient.errors.HttpError as err:
132      raise utils.GcpApiError(err) from err
133  return services