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(match.group('project'),
110                                   match.group('network'))
111      else:
112        return network.get_network(self.project_id, self.authorized_network)
113
114    return network.get_network(self.project_id, 'default')

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

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

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

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

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

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

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

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