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)
Represents an Interconnect. https://cloud.google.com/compute/docs/reference/rest/v1/interconnects
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'
@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]
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 links = [] 115 if isinstance(response, dict): 116 # Handle the case when 'response' is a dictionary 117 links = [ 118 Interconnect(project_id, name) for name in response.get('items', []) 119 ] 120 elif isinstance(response, list): 121 # Handle the case when 'response' is a list 122 links = [Interconnect(project_id, name) for name in response] 123 return links
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'
Represents an Interconnect. https://cloud.google.com/compute/docs/reference/rest/v1/interconnectAttachments
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'
@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