gcpdiag.queries.apigee

33class ApigeeEnvironment(models.Resource):
34  """Represents an Apigee Environment
35    https://cloud.google.com/apigee/docs/reference/apis/apigee/rest/v1/organizations.environments#Environment
36    """
37
38  def __init__(self, apigee_org, env_name: str):
39    super().__init__(project_id=apigee_org.project_id)
40    self.org_name = apigee_org.name
41    self.name = env_name
42
43  @property
44  def full_path(self) -> str:
45    return f'organizations/{self.org_name}/environments/{self.name}'

 48class ApigeeOrganization(models.Resource):
 49  """Represents an Apigee Organization
 50    https://cloud.google.com/apigee/docs/reference/apis/apigee/rest/v1/organizations#Organization
 51    """
 52  _resource_data: Optional[dict]
 53  _environments: List[ApigeeEnvironment]
 54
 55  # It is possible to create an Apigee organization resource in another GCP project,
 56  # set the resource_data as optional to avoid permission issues while accessing another GCP project
 57  def __init__(self,
 58               project_id: str,
 59               org_name: str,
 60               resource_data: Optional[dict] = None):
 61    super().__init__(project_id=project_id)
 62    self._resource_data = resource_data
 63    self.name = org_name
 64
 65  @property
 66  def full_path(self) -> str:
 67    return f'organizations/{self.name}'
 68
 69  @property
 70  def environments(self) -> Iterable[ApigeeEnvironment]:
 71    if self._resource_data is None:
 72      return []
 73
 74    return [
 75        ApigeeEnvironment(self, env)
 76        for env in self._resource_data.get('environments', [])
 77    ]
 78
 79  @property
 80  def runtime_type(self) -> str:
 81    if self._resource_data is None:
 82      return ''
 83    return self._resource_data['runtimeType']
 84
 85  @property
 86  def runtime_database_encryption_key_name(self) -> str:
 87    if self._resource_data is None:
 88      return ''
 89
 90    return self._resource_data.get('runtimeDatabaseEncryptionKeyName', '')
 91
 92  @property
 93  def authorized_network(self) -> str:
 94    if self._resource_data is None:
 95      return ''
 96
 97    return self._resource_data.get('authorizedNetwork', '')
 98
 99  @property
100  def network(self) -> network.Network:
101    if self.authorized_network:
102      match = re.match(
103          r'projects/(?P<project>[^/]+)/([^/]+)/networks/(?P<network>[^/]+)$',
104          self.authorized_network)
105      # Check whether the authorized network is a shared VPC network
106      # A shared VPC network is using following format:
107      # `projects/{host-project-id}/{region}/networks/{network-name}`
108      if match:
109        return network.get_network(
110            match.group('project'), match.group('network'),
111            models.Context(project_id=match.group('project')))
112      else:
113        return network.get_network(self.project_id, self.authorized_network,
114                                   models.Context(project_id=self.project_id))
115
116    return network.get_network(self.project_id, 'default',
117                               models.Context(project_id=self.project_id))

120class EnvironmentGroup(models.Resource):
121  """Represents an Apigee Environment Group
122    https://cloud.google.com/apigee/docs/reference/apis/apigee/rest/v1/organizations.envgroups#resource:-environmentgroup
123    """
124  _resource_data: dict
125
126  def __init__(self, apigee_org: ApigeeOrganization, resource_data):
127    super().__init__(project_id=apigee_org.project_id)
128    self._resource_data = resource_data
129    self.org_name = apigee_org.name
130
131  @property
132  def name(self) -> str:
133    return self._resource_data['name']
134
135  @property
136  def full_path(self) -> str:
137    return f'organizations/{self.org_name}/envgroups/{self.name}'
138
139  @property
140  def host_names(self) -> List[str]:
141    return self._resource_data['hostnames']

144class ApigeeInstance(models.Resource):
145  """Represents an Apigee Runtime Instance
146    https://cloud.google.com/apigee/docs/reference/apis/apigee/rest/v1/organizations.instances#Instance
147    """
148  _resource_data: dict
149
150  def __init__(self, apigee_org: ApigeeOrganization, resource_data):
151    super().__init__(project_id=apigee_org.project_id)
152    self._resource_data = resource_data
153    self.org_name = apigee_org.name
154
155  @property
156  def name(self) -> str:
157    return self._resource_data['name']
158
159  @property
160  def full_path(self) -> str:
161    return f'organizations/{self.org_name}/instances/{self.name}'
162
163  @property
164  def disk_encryption_key_name(self) -> str:
165    return self._resource_data.get('diskEncryptionKeyName', '')
166
167  @property
168  def host(self) -> str:
169    return self._resource_data.get('host', '')
170
171  @property
172  def location(self) -> str:
173    return self._resource_data.get('location', '')

176@caching.cached_api_call
177def get_org(context: models.Context) -> Optional[ApigeeOrganization]:
178  """Get Apigee organizations matching the GCP Project Id"""
179  if not apis.is_enabled(context.project_id, 'apigee'):
180    return None
181  apigee_api = apis.get_api('apigee', 'v1', context.project_id)
182  # Apigee Organization : GCP Project = 1 : 1
183  org_list_query = apigee_api.organizations().list(parent='organizations')
184  try:
185    resp = org_list_query.execute(num_retries=config.API_RETRIES)
186    if 'organizations' not in resp:
187      return None
188    for resp_o in resp['organizations']:
189      if 'organization' not in resp_o or 'projectId' not in resp_o:
190        raise RuntimeError('missing data in organizations.list response')
191      if context.project_id == resp_o['projectId']:
192        org_name = resp_o['organization']
193        get_org_query = apigee_api.organizations().get(
194            name=f'organizations/{org_name}')
195        get_org_resp = get_org_query.execute(num_retries=config.API_RETRIES)
196        return ApigeeOrganization(context.project_id, resp_o['organization'],
197                                  get_org_resp)
198  except googleapiclient.errors.HttpError as err:
199    raise GcpApiError(err) from err
200  return None

203@caching.cached_api_call
204def get_envgroups(
205    apigee_org: ApigeeOrganization) -> Mapping[str, EnvironmentGroup]:
206  """Get Environment group list by organization name, caching the result."""
207  envgroups: Dict[str, EnvironmentGroup] = {}
208  apigee_api = apis.get_api('apigee', 'v1')
209
210  request = apigee_api.organizations().envgroups().list(
211      parent=f'organizations/{apigee_org.name}')
212  for envgroup in apis_utils.list_all(
213      request,
214      next_function=apigee_api.organizations().envgroups().list_next,
215      response_keyword='environmentGroups'):
216    envgroups[envgroup['name']] = EnvironmentGroup(apigee_org=apigee_org,
217                                                   resource_data=envgroup)
218  return envgroups

221@caching.cached_api_call
222def get_envgroups_attachments(envgroup_name: str) -> List[str]:
223  """Get Environment group attachments by environment group name, caching the result."""
224  environments: List[str] = []
225  apigee_api = apis.get_api('apigee', 'v1')
226
227  request = apigee_api.organizations().envgroups().attachments().list(
228      parent=envgroup_name)
229  for attachments in apis_utils.list_all(
230      request,
231      next_function=apigee_api.organizations().envgroups().attachments(
232      ).list_next,
233      response_keyword='environmentGroupAttachments'):
234    environments.append(attachments['environment'])
235  return environments

238@caching.cached_api_call
239def get_instances(
240    apigee_org: ApigeeOrganization) -> Mapping[str, ApigeeInstance]:
241  """Get instance list from Apigee Organization, caching the result."""
242  instances: Dict[str, ApigeeInstance] = {}
243  # Not supported for Apigee hybrid.
244  if apigee_org.runtime_type == 'HYBRID':
245    return instances
246
247  apigee_api = apis.get_api('apigee', 'v1')
248  request = apigee_api.organizations().instances().list(
249      parent=f'organizations/{apigee_org.name}')
250  for instance in apis_utils.list_all(
251      request,
252      next_function=apigee_api.organizations().instances().list_next,
253      response_keyword='instances'):
254    instances[instance['name']] = ApigeeInstance(apigee_org=apigee_org,
255                                                 resource_data=instance)
256  return instances

259@caching.cached_api_call
260def get_instances_attachments(instance_name: str) -> List[str]:
261  """Get instance attachments by instance name, caching the result."""
262  environments: List[str] = []
263  if not instance_name:
264    return environments
265
266  apigee_api = apis.get_api('apigee', 'v1')
267  request = apigee_api.organizations().instances().attachments().list(
268      parent=instance_name)
269  for attachments in apis_utils.list_all(request,
270                                         next_function=apigee_api.organizations(
271                                         ).instances().attachments().list_next,
272                                         response_keyword='attachments'):
273    environments.append(attachments['environment'])
274  return environments

277@functools.lru_cache()
278def get_network_bridge_instance_groups(
279    project_id: str) -> List[gce.ManagedInstanceGroup]:
280  """Get a list of managed instance groups used by Apigee for routing purposes."""
281  migs: List[gce.ManagedInstanceGroup] = []
282  for m in gce.get_region_managed_instance_groups(
283      models.Context(project_id=project_id)).values():
284    if m.template.get_metadata('startup-script-url') == MIG_STARTUP_SCRIPT_URL:
285      migs.append(m)
286  return migs