gcpdiag.queries.apigee

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

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

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
125  _resource_data: dict
126
127  def __init__(self, apigee_org: ApigeeOrganization, resource_data):
128    super().__init__(project_id=apigee_org.project_id)
129    self._resource_data = resource_data
130    self.org_name = apigee_org.name
131
132  @property
133  def name(self) -> str:
134    return self._resource_data['name']
135
136  @property
137  def full_path(self) -> str:
138    return f'organizations/{self.org_name}/envgroups/{self.name}'
139
140  @property
141  def host_names(self) -> List[str]:
142    return self._resource_data['hostnames']

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

178@caching.cached_api_call
179def get_org(context: models.Context) -> Optional[ApigeeOrganization]:
180  """Get Apigee organizations matching the GCP Project Id"""
181  if not apis.is_enabled(context.project_id, 'apigee'):
182    return None
183  apigee_api = apis.get_api('apigee', 'v1', context.project_id)
184  # Apigee Organization : GCP Project = 1 : 1
185  org_list_query = apigee_api.organizations().list(parent='organizations')
186  try:
187    resp = org_list_query.execute(num_retries=config.API_RETRIES)
188    if 'organizations' not in resp:
189      return None
190    for resp_o in resp['organizations']:
191      if 'organization' not in resp_o or 'projectId' not in resp_o:
192        raise RuntimeError('missing data in organizations.list response')
193      if context.project_id == resp_o['projectId']:
194        org_name = resp_o['organization']
195        get_org_query = apigee_api.organizations().get(name=f'organizations/{org_name}')
196        get_org_resp = get_org_query.execute(num_retries=config.API_RETRIES)
197        return ApigeeOrganization(context.project_id, resp_o['organization'], 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(apigee_org: ApigeeOrganization) -> Mapping[str, EnvironmentGroup]:
205  """Get Environment group list by organization name, caching the result."""
206  envgroups: Dict[str, EnvironmentGroup] = {}
207  apigee_api = apis.get_api('apigee', 'v1')
208
209  request = apigee_api.organizations().envgroups().list(parent=f'organizations/{apigee_org.name}')
210  for envgroup in apis_utils.list_all(
211    request,
212    next_function=apigee_api.organizations().envgroups().list_next,
213    response_keyword='environmentGroups',
214  ):
215    envgroups[envgroup['name']] = EnvironmentGroup(apigee_org=apigee_org, resource_data=envgroup)
216  return envgroups

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

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

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

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