gcpdiag.queries.interconnect

Queries related to interconnects.
class Interconnect(gcpdiag.models.Resource):
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)
Interconnect(project_id, resource_data)
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 = ''
name: str
39  @property
40  def name(self) -> str:
41    return self._resource_data['name']
under_maintenance: bool
43  @property
44  def under_maintenance(self) -> bool:
45    return self._resource_data['state'] == 'UNDER_MAINTENANCE'
id: str
47  @property
48  def id(self) -> str:
49    return self._resource_data['id']
full_path: str
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}'

Returns the full path of this resource.

Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'

short_path: str
60  @property
61  def short_path(self) -> str:
62    path = self.project_id + '/' + self.name
63    return path

Returns the short name for this resource.

Note that it isn't clear from this name what kind of resource it is.

Example: 'gke1'

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

Returns the full path of this resource.

Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'

short_path: str
163  @property
164  def short_path(self) -> str:
165    path = self.project_id + '/' + self.name
166    return path

Returns the short name for this resource.

Note that it isn't clear from this name what kind of resource it is.

Example: 'gke1'

type: str
172  @property
173  def type(self) -> str:
174    return self._resource_data['type']
interconnect: str
176  @property
177  def interconnect(self) -> str:
178    if not self._interconnect:
179      self._interconnect = self._resource_data['interconnect'].split('/')[-1]
180    return self._interconnect
router: str
182  @property
183  def router(self) -> str:
184    return self._resource_data['router'].split('/')[-1]
region: str
186  @property
187  def region(self) -> str:
188    return self._resource_data['region'].split('/')[-1]
ead: str
190  @property
191  def ead(self) -> str:
192    if not self._ead:
193      interconnect_obj = get_interconnect(self.project_id, self.interconnect)
194      self._ead = interconnect_obj.ead
195    return self._ead
metro: str
197  @property
198  def metro(self) -> str:
199    return _metro(self.ead)
legacy_dataplane: bool
201  @property
202  def legacy_dataplane(self) -> bool:
203    if 'dataplaneVersion' not in self._resource_data:
204      self._resource_data['dataplaneVersion'] = {}
205
206    return self._resource_data['dataplaneVersion'] != 2
defunct_state: bool
208  @property
209  def defunct_state(self) -> bool:
210    return self._resource_data['state'] == 'DEFUNCT'
@caching.cached_api_call(in_memory=True)
def get_vlan_attachment( project_id: str, region: str, vlan_attachment: str) -> VlanAttachment:
213@caching.cached_api_call(in_memory=True)
214def get_vlan_attachment(project_id: str, region: str,
215                        vlan_attachment: str) -> VlanAttachment:
216  logging.info('fetching vlan attachment: %s', vlan_attachment)
217  compute = apis.get_api('compute', 'v1', project_id)
218  request = compute.interconnectAttachments().get(
219      project=project_id, region=region, interconnectAttachment=vlan_attachment)
220  response = request.execute(num_retries=config.API_RETRIES)
221  return VlanAttachment(project_id, response)
@caching.cached_api_call(in_memory=True)
def get_vlan_attachments(project_id: str) -> List[VlanAttachment]:
224@caching.cached_api_call(in_memory=True)
225def get_vlan_attachments(project_id: str) -> List[VlanAttachment]:
226  logging.info('fetching vlan attachments')
227  compute = apis.get_api('compute', 'v1', project_id)
228  attachments = []
229  request = compute.interconnectAttachments().aggregatedList(project=project_id)
230  response = request.execute(num_retries=config.API_RETRIES)
231  attachments_by_regions = response['items']
232  for _, data_ in attachments_by_regions.items():
233    if 'interconnectAttachments' not in data_:
234      continue
235    attachments.extend([
236        VlanAttachment(project_id, va)
237        for va in data_['interconnectAttachments']
238    ])
239  return attachments