gcpdiag.queries.interconnect

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

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

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

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

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

212@caching.cached_api_call(in_memory=True)
213def get_vlan_attachment(project_id: str, region: str,
214                        vlan_attachment: str) -> VlanAttachment:
215  logging.info('fetching vlan attachment: %s', vlan_attachment)
216  compute = apis.get_api('compute', 'v1', project_id)
217  request = compute.interconnectAttachments().get(
218      project=project_id, region=region, interconnectAttachment=vlan_attachment)
219  response = request.execute(num_retries=config.API_RETRIES)
220  return VlanAttachment(project_id, response)

223@caching.cached_api_call(in_memory=True)
224def get_vlan_attachments(project_id: str) -> List[VlanAttachment]:
225  logging.info('fetching vlan attachments')
226  compute = apis.get_api('compute', 'v1', project_id)
227  attachments = []
228  request = compute.interconnectAttachments().aggregatedList(project=project_id)
229  response = request.execute(num_retries=config.API_RETRIES)
230  attachments_by_regions = response['items']
231  for _, data_ in attachments_by_regions.items():
232    if 'interconnectAttachments' not in data_:
233      continue
234    attachments.extend([
235        VlanAttachment(project_id, va)
236        for va in data_['interconnectAttachments']
237    ])
238  return attachments