gcpdiag.queries.interconnect

27class Interconnect(models.Resource):
28  """Represents an Interconnect.
29 https://cloud.google.com/compute/docs/reference/rest/v1/interconnects
30 """
31  _resource_data: dict
32  _ead: str
33  _attachments: List[str]
34
35  def __init__(self, project_id, resource_data):
36    super().__init__(project_id=project_id)
37    self._resource_data = resource_data
38    self._attachments = []
39    self._ead = ''
40
41  @property
42  def name(self) -> str:
43    return self._resource_data['name']
44
45  @property
46  def under_maintenance(self) -> bool:
47    return self._resource_data['state'] == 'UNDER_MAINTENANCE'
48
49  @property
50  def id(self) -> str:
51    return self._resource_data['id']
52
53  @property
54  def full_path(self) -> str:
55    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)',
56                      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 = [
75          x.split('/')[-1]
76          for x in self._resource_data['interconnectAttachments']
77      ]
78    return self._attachments
79
80  @property
81  def ead(self) -> str:
82    if not self._ead:
83      self._ead = self._resource_data['location'].split('/')[-1]
84    return self._ead
85
86  @property
87  def metro(self) -> str:
88    return _metro(self.ead)

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

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

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

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

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

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