gcpdiag.queries.interconnect

27class Interconnect(models.Resource):
28  """Represents an Interconnect.
29  https://cloud.google.com/compute/docs/reference/rest/v1/interconnects
30  """
31
32  _resource_data: dict
33  _ead: str
34  _attachments: List[str]
35
36  def __init__(self, project_id, resource_data):
37    super().__init__(project_id=project_id)
38    self._resource_data = resource_data
39    self._attachments = []
40    self._ead = ''
41
42  @property
43  def name(self) -> str:
44    return self._resource_data['name']
45
46  @property
47  def under_maintenance(self) -> bool:
48    return self._resource_data['state'] == 'UNDER_MAINTENANCE'
49
50  @property
51  def id(self) -> str:
52    return self._resource_data['id']
53
54  @property
55  def full_path(self) -> str:
56    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', self.self_link)
57    if result:
58      return result.group(1)
59    else:
60      return f'>> {self.self_link}'
61
62  @property
63  def short_path(self) -> str:
64    path = self.project_id + '/' + self.name
65    return path
66
67  @property
68  def self_link(self) -> str:
69    return self._resource_data['selfLink']
70
71  @property
72  def attachments(self) -> List[str]:
73    if not self._attachments:
74      self._attachments = [x.split('/')[-1] for x in self._resource_data['interconnectAttachments']]
75    return self._attachments
76
77  @property
78  def ead(self) -> str:
79    if not self._ead:
80      self._ead = self._resource_data['location'].split('/')[-1]
81    return self._ead
82
83  @property
84  def metro(self) -> str:
85    return _metro(self.ead)

88@caching.cached_api_call(in_memory=True)
89def get_interconnect(project_id: str, interconnect_name: str) -> Interconnect:
90  logging.debug('fetching interconnect: %s', interconnect_name)
91  compute = apis.get_api('compute', 'v1', project_id)
92  request = compute.interconnects().get(project=project_id, interconnect=interconnect_name)
93  response = request.execute(num_retries=config.API_RETRIES)
94  return Interconnect(project_id, response)

 97@caching.cached_api_call(in_memory=True)
 98def get_interconnects(project_id: str) -> List[Interconnect]:
 99  logging.info('fetching interconnects')
100  compute = apis.get_api('compute', 'v1', project_id)
101  request = compute.interconnects().list(project=project_id)
102  response = request.execute(num_retries=config.API_RETRIES)
103  return [Interconnect(project_id, link) for link in response]

106@caching.cached_api_call(in_memory=True)
107def get_links(project_id: str) -> List[Interconnect]:
108  logging.info('fetching interconnects')
109  compute = apis.get_api('compute', 'v1', project_id)
110  request = compute.interconnects().list(project=project_id)
111  response = request.execute(num_retries=config.API_RETRIES)
112  links = []
113  if isinstance(response, dict):
114    # Handle the case when 'response' is a dictionary
115    links = [Interconnect(project_id, name) for name in response.get('items', [])]
116  elif isinstance(response, list):
117    # Handle the case when 'response' is a list
118    links = [Interconnect(project_id, name) for name in response]
119  return links

126class VlanAttachment(models.Resource):
127  """Represents an Interconnect.
128  https://cloud.google.com/compute/docs/reference/rest/v1/interconnectAttachments
129  """
130
131  _resource_data: dict
132  _type: str
133  _interconnect: str
134  _ead: str
135
136  def __init__(self, project_id, resource_data):
137    super().__init__(project_id=project_id)
138    self._resource_data = resource_data
139    self._type = ''
140    self._interconnect = ''
141    self._ead = ''
142
143  @property
144  def name(self) -> str:
145    return self._resource_data['name']
146
147  @property
148  def id(self) -> str:
149    return self._resource_data['id']
150
151  @property
152  def full_path(self) -> str:
153    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', self.self_link)
154    if result:
155      return result.group(1)
156    else:
157      return f'>> {self.self_link}'
158
159  @property
160  def short_path(self) -> str:
161    path = self.project_id + '/' + self.name
162    return path
163
164  @property
165  def self_link(self) -> str:
166    return self._resource_data['selfLink']
167
168  @property
169  def type(self) -> str:
170    return self._resource_data['type']
171
172  @property
173  def interconnect(self) -> str:
174    if not self._interconnect:
175      self._interconnect = self._resource_data['interconnect'].split('/')[-1]
176    return self._interconnect
177
178  @property
179  def router(self) -> str:
180    return self._resource_data['router'].split('/')[-1]
181
182  @property
183  def region(self) -> str:
184    return self._resource_data['region'].split('/')[-1]
185
186  @property
187  def mtu(self) -> int:
188    if 'mtu' in self._resource_data:
189      return self._resource_data['mtu']
190    else:
191      return DEFAULT_MTU
192
193  @property
194  def ipv4address(self) -> str:
195    return self._resource_data['cloudRouterIpAddress'].split('/')[0]
196
197  @property
198  def remoteip(self) -> str:
199    return self._resource_data['customerRouterIpAddress'].split('/')[0]
200
201  @property
202  def ead(self) -> str:
203    if not self._ead:
204      interconnect_obj = get_interconnect(self.project_id, self.interconnect)
205      self._ead = interconnect_obj.ead
206    return self._ead
207
208  @property
209  def metro(self) -> str:
210    return _metro(self.ead)
211
212  @property
213  def legacy_dataplane(self) -> bool:
214    if 'dataplaneVersion' not in self._resource_data:
215      self._resource_data['dataplaneVersion'] = {}
216
217    return self._resource_data['dataplaneVersion'] != 2
218
219  @property
220  def defunct_state(self) -> bool:
221    return self._resource_data['state'] == 'DEFUNCT'

224@caching.cached_api_call(in_memory=True)
225def get_vlan_attachment(project_id: str, region: str, vlan_attachment: str) -> VlanAttachment:
226  logging.debug('fetching vlan attachment: %s', vlan_attachment)
227  compute = apis.get_api('compute', 'v1', project_id)
228  request = compute.interconnectAttachments().get(
229    project=project_id, region=region, interconnectAttachment=vlan_attachment
230  )
231  response = request.execute(num_retries=config.API_RETRIES)
232  return VlanAttachment(project_id, response)

235@caching.cached_api_call(in_memory=True)
236def get_vlan_attachments(project_id: str) -> List[VlanAttachment]:
237  logging.info('fetching vlan attachments')
238  compute = apis.get_api('compute', 'v1', project_id)
239  attachments = []
240  request = compute.interconnectAttachments().aggregatedList(project=project_id)
241  response = request.execute(num_retries=config.API_RETRIES)
242  attachments_by_regions = response['items']
243  for _, data_ in attachments_by_regions.items():
244    if 'interconnectAttachments' not in data_:
245      continue
246    attachments.extend([VlanAttachment(project_id, va) for va in data_['interconnectAttachments']])
247  return attachments