gcpdiag.queries.lb
28class LoadBalancerType(Enum): 29 """Load balancer type.""" 30 31 LOAD_BALANCER_TYPE_UNSPECIFIED = 0 32 EXTERNAL_PASSTHROUGH_LB = 1 33 INTERNAL_PASSTHROUGH_LB = 2 34 TARGET_POOL_LB = 3 # deprecated but customers still have them 35 GLOBAL_EXTERNAL_PROXY_NETWORK_LB = 4 # envoy based proxy lb 36 REGIONAL_INTERNAL_PROXY_NETWORK_LB = 5 37 REGIONAL_EXTERNAL_PROXY_NETWORK_LB = 6 38 CROSS_REGION_INTERNAL_PROXY_NETWORK_LB = 7 39 CLASSIC_PROXY_NETWORK_LB = 8 40 GLOBAL_EXTERNAL_APPLICATION_LB = 9 # envoy based application lb 41 REGIONAL_INTERNAL_APPLICATION_LB = 10 42 REGIONAL_EXTERNAL_APPLICATION_LB = 11 43 CROSS_REGION_INTERNAL_APPLICATION_LB = 12 44 CLASSIC_APPLICATION_LB = 13
Load balancer type.
47def get_load_balancer_type_name(lb_type: LoadBalancerType) -> str: 48 """Returns a human-readable name for the given load balancer type.""" 49 50 type_names = { 51 LoadBalancerType.LOAD_BALANCER_TYPE_UNSPECIFIED: 52 'Unspecified', 53 LoadBalancerType.EXTERNAL_PASSTHROUGH_LB: 54 ('External Passthrough Network Load Balancer'), 55 LoadBalancerType.INTERNAL_PASSTHROUGH_LB: 56 ('Internal Passthrough Network Load Balancer'), 57 LoadBalancerType.TARGET_POOL_LB: 58 'Target Pool Network Load Balancer', 59 LoadBalancerType.GLOBAL_EXTERNAL_PROXY_NETWORK_LB: 60 ('Global External Proxy Network Load Balancer'), 61 LoadBalancerType.REGIONAL_INTERNAL_PROXY_NETWORK_LB: 62 ('Regional Internal Proxy Network Load Balancer'), 63 LoadBalancerType.REGIONAL_EXTERNAL_PROXY_NETWORK_LB: 64 ('Regional External Proxy Network Load Balancer'), 65 LoadBalancerType.CROSS_REGION_INTERNAL_PROXY_NETWORK_LB: 66 ('Cross-Region Internal Proxy Network Load Balancer'), 67 LoadBalancerType.CLASSIC_PROXY_NETWORK_LB: 68 ('Classic Proxy Network Load Balancer'), 69 LoadBalancerType.GLOBAL_EXTERNAL_APPLICATION_LB: 70 ('Global External Application Load Balancer'), 71 LoadBalancerType.REGIONAL_INTERNAL_APPLICATION_LB: 72 ('Regional Internal Application Load Balancer'), 73 LoadBalancerType.REGIONAL_EXTERNAL_APPLICATION_LB: 74 ('Regional External Application Load Balancer'), 75 LoadBalancerType.CROSS_REGION_INTERNAL_APPLICATION_LB: 76 ('Cross-Region Internal Application Load Balancer'), 77 LoadBalancerType.CLASSIC_APPLICATION_LB: 78 ('Classic Application Load Balancer'), 79 } 80 return type_names.get(lb_type, 'Unspecified')
Returns a human-readable name for the given load balancer type.
83def get_load_balancer_type( 84 load_balancing_scheme: str, 85 scope: str, 86 layer: Literal['application', 'network'], 87 backend_service_based: bool = True, 88) -> LoadBalancerType: 89 if load_balancing_scheme == 'EXTERNAL': 90 if not scope or scope == 'global': 91 if layer == 'application': 92 return LoadBalancerType.CLASSIC_APPLICATION_LB 93 else: 94 return LoadBalancerType.CLASSIC_PROXY_NETWORK_LB 95 else: 96 return (LoadBalancerType.EXTERNAL_PASSTHROUGH_LB 97 if backend_service_based else LoadBalancerType.TARGET_POOL_LB) 98 elif load_balancing_scheme == 'INTERNAL': 99 return LoadBalancerType.INTERNAL_PASSTHROUGH_LB 100 elif load_balancing_scheme == 'INTERNAL_MANAGED': 101 if not scope or scope == 'global': 102 if layer == 'application': 103 return LoadBalancerType.CROSS_REGION_INTERNAL_APPLICATION_LB 104 else: 105 return LoadBalancerType.CROSS_REGION_INTERNAL_PROXY_NETWORK_LB 106 else: 107 if layer == 'application': 108 return LoadBalancerType.REGIONAL_INTERNAL_APPLICATION_LB 109 else: 110 return LoadBalancerType.REGIONAL_INTERNAL_PROXY_NETWORK_LB 111 elif load_balancing_scheme == 'EXTERNAL_MANAGED': 112 if not scope or scope == 'global': 113 if layer == 'application': 114 return LoadBalancerType.GLOBAL_EXTERNAL_APPLICATION_LB 115 else: 116 return LoadBalancerType.GLOBAL_EXTERNAL_PROXY_NETWORK_LB 117 else: 118 if layer == 'application': 119 return LoadBalancerType.REGIONAL_EXTERNAL_APPLICATION_LB 120 else: 121 return LoadBalancerType.REGIONAL_EXTERNAL_PROXY_NETWORK_LB 122 return LoadBalancerType.LOAD_BALANCER_TYPE_UNSPECIFIED
125def normalize_url(url: str) -> str: 126 """Returns normalized url.""" 127 result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', url) 128 if result: 129 return result.group(1) 130 else: 131 return ''
Returns normalized url.
134class BackendServices(models.Resource): 135 """A Backend Service resource.""" 136 137 _resource_data: dict 138 _type: str 139 140 def __init__(self, project_id, resource_data): 141 super().__init__(project_id=project_id) 142 self._resource_data = resource_data 143 144 @property 145 def name(self) -> str: 146 return self._resource_data['name'] 147 148 @property 149 def id(self) -> str: 150 return self._resource_data['id'] 151 152 @property 153 def full_path(self) -> str: 154 result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', 155 self.self_link) 156 if result: 157 return result.group(1) 158 else: 159 return f'>> {self.self_link}' 160 161 @property 162 def short_path(self) -> str: 163 path = self.project_id + '/' + self.name 164 return path 165 166 @property 167 def self_link(self) -> str: 168 return self._resource_data['selfLink'] 169 170 @property 171 def session_affinity(self) -> str: 172 return self._resource_data.get('sessionAffinity', 'NONE') 173 174 @property 175 def timeout_sec(self) -> int: 176 return self._resource_data.get('timeoutSec', None) 177 178 @property 179 def locality_lb_policy(self) -> str: 180 return self._resource_data.get('localityLbPolicy', 'ROUND_ROBIN') 181 182 @property 183 def is_enable_cdn(self) -> str: 184 return self._resource_data.get('enableCDN', False) 185 186 @property 187 def draining_timeout_sec(self) -> int: 188 return self._resource_data.get('connectionDraining', 189 {}).get('drainingTimeoutSec', 0) 190 191 @property 192 def load_balancing_scheme(self) -> str: 193 return self._resource_data.get('loadBalancingScheme', None) 194 195 @property 196 def health_check(self) -> str: 197 health_check_url = self._resource_data['healthChecks'][0] 198 matches = re.search(r'/([^/]+)$', health_check_url) 199 if matches: 200 healthcheck_name = matches.group(1) 201 return healthcheck_name 202 else: 203 return '' 204 205 @property 206 def backends(self) -> List[dict]: 207 return self._resource_data.get('backends', []) 208 209 @property 210 def region(self): 211 try: 212 url = self._resource_data.get('region') 213 if url is not None: 214 match = re.search(r'/([^/]+)/?$', url) 215 if match is not None: 216 region = match.group(1) 217 return region 218 else: 219 return None 220 except KeyError: 221 return None 222 223 @property 224 def protocol(self) -> str: 225 return self._resource_data.get('protocol', None) 226 227 @property 228 def used_by_refs(self) -> List[str]: 229 used_by = [] 230 for x in self._resource_data.get('usedBy', []): 231 reference = x.get('reference') 232 if reference: 233 match = re.match(r'https://www.googleapis.com/compute/v1/(.*)', 234 reference) 235 if match: 236 used_by.append(match.group(1)) 237 return used_by 238 239 @property 240 def load_balancer_type(self) -> LoadBalancerType: 241 application_protocols = ['HTTP', 'HTTPS', 'HTTP2'] 242 return get_load_balancer_type( 243 self.load_balancing_scheme, 244 self.region, 245 'application' if self.protocol in application_protocols else 'network', 246 backend_service_based=True, 247 )
A Backend Service resource.
152 @property 153 def full_path(self) -> str: 154 result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', 155 self.self_link) 156 if result: 157 return result.group(1) 158 else: 159 return f'>> {self.self_link}'
Returns the full path of this resource.
Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'
161 @property 162 def short_path(self) -> str: 163 path = self.project_id + '/' + self.name 164 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'
227 @property 228 def used_by_refs(self) -> List[str]: 229 used_by = [] 230 for x in self._resource_data.get('usedBy', []): 231 reference = x.get('reference') 232 if reference: 233 match = re.match(r'https://www.googleapis.com/compute/v1/(.*)', 234 reference) 235 if match: 236 used_by.append(match.group(1)) 237 return used_by
239 @property 240 def load_balancer_type(self) -> LoadBalancerType: 241 application_protocols = ['HTTP', 'HTTPS', 'HTTP2'] 242 return get_load_balancer_type( 243 self.load_balancing_scheme, 244 self.region, 245 'application' if self.protocol in application_protocols else 'network', 246 backend_service_based=True, 247 )
250@caching.cached_api_call(in_memory=True) 251def get_backend_services(project_id: str) -> List[BackendServices]: 252 logging.debug('fetching Backend Services: %s', project_id) 253 compute = apis.get_api('compute', 'v1', project_id) 254 backend_services = [] 255 request = compute.backendServices().aggregatedList(project=project_id) 256 response = request.execute(num_retries=config.API_RETRIES) 257 backend_services_by_region = response['items'] 258 for _, data_ in backend_services_by_region.items(): 259 if 'backendServices' not in data_: 260 continue 261 backend_services.extend([ 262 BackendServices(project_id, backend_service) 263 for backend_service in data_['backendServices'] 264 ]) 265 return backend_services
268@caching.cached_api_call(in_memory=True) 269def get_backend_service(project_id: str, 270 backend_service_name: str, 271 region: str = None) -> BackendServices: 272 """Returns instance object matching backend service name and region""" 273 compute = apis.get_api('compute', 'v1', project_id) 274 try: 275 if not region or region == 'global': 276 request = compute.backendServices().get( 277 project=project_id, backendService=backend_service_name) 278 else: 279 request = compute.regionBackendServices().get( 280 project=project_id, 281 region=region, 282 backendService=backend_service_name) 283 284 response = request.execute(num_retries=config.API_RETRIES) 285 return BackendServices(project_id, resource_data=response) 286 except googleapiclient.errors.HttpError as err: 287 raise utils.GcpApiError(err) from err
Returns instance object matching backend service name and region
290def get_backend_service_by_self_link( 291 backend_service_self_link: str,) -> Optional[BackendServices]: 292 backend_service_name = backend_service_self_link.split('/')[-1] 293 backend_service_scope = backend_service_self_link.split('/')[-3] 294 match = re.match(r'projects/([^/]+)/', backend_service_self_link) 295 if not match: 296 return None 297 project_id = match.group(1) 298 return get_backend_service(project_id, backend_service_name, 299 backend_service_scope)
302class BackendHealth: 303 """A Backend Service resource.""" 304 305 _resource_data: dict 306 307 def __init__(self, resource_data, group): 308 self._resource_data = resource_data 309 self._group = group 310 311 @property 312 def instance(self) -> str: 313 return self._resource_data['instance'] 314 315 @property 316 def group(self) -> str: 317 return self._group 318 319 @property 320 def health_state(self) -> str: 321 return self._resource_data.get('healthState', 'UNHEALTHY')
A Backend Service resource.
324@caching.cached_api_call(in_memory=True) 325def get_backend_service_health( 326 project_id: str, 327 backend_service_name: str, 328 backend_service_region: str = None, 329) -> List[BackendHealth]: 330 """Returns health data for backend service.""" 331 try: 332 backend_service = get_backend_service(project_id, backend_service_name, 333 backend_service_region) 334 except googleapiclient.errors.HttpError: 335 return [] 336 337 backend_heath_statuses: List[BackendHealth] = [] 338 339 compute = apis.get_api('compute', 'v1', project_id) 340 341 for backend in backend_service.backends: 342 group = backend['group'] 343 if not backend_service.region: 344 response = (compute.backendServices().getHealth( 345 project=project_id, 346 backendService=backend_service.name, 347 body={ 348 'group': group 349 }, 350 ).execute(num_retries=config.API_RETRIES)) 351 # None is returned when backend type doesn't support health check 352 if response is not None: 353 for health_status in response.get('healthStatus', []): 354 backend_heath_statuses.append(BackendHealth(health_status, group)) 355 else: 356 response = (compute.regionBackendServices().getHealth( 357 project=project_id, 358 region=backend_service.region, 359 backendService=backend_service.name, 360 body={ 361 'group': group 362 }, 363 ).execute(num_retries=config.API_RETRIES)) 364 if response is not None: 365 for health_status in response.get('healthStatus', []): 366 backend_heath_statuses.append(BackendHealth(health_status, group)) 367 368 return backend_heath_statuses
Returns health data for backend service.
371class SslCertificate(models.Resource): 372 """A SSL Certificate resource.""" 373 374 _resource_data: dict 375 _type: str 376 377 def __init__(self, project_id, resource_data): 378 super().__init__(project_id=project_id) 379 self._resource_data = resource_data 380 381 @property 382 def name(self) -> str: 383 return self._resource_data['name'] 384 385 @property 386 def id(self) -> str: 387 return self._resource_data['id'] 388 389 @property 390 def full_path(self) -> str: 391 result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', 392 self.self_link) 393 if result: 394 return result.group(1) 395 else: 396 return f'>> {self.self_link}' 397 398 @property 399 def self_link(self) -> str: 400 return self._resource_data['selfLink'] 401 402 @property 403 def type(self) -> str: 404 return self._resource_data.get('type', 'SELF_MANAGED') 405 406 @property 407 def status(self) -> str: 408 return self._resource_data.get('managed', {}).get('status') 409 410 @property 411 def domains(self) -> List[str]: 412 return self._resource_data.get('managed', {}).get('domains', []) 413 414 @property 415 def domain_status(self) -> Dict[str, str]: 416 return self._resource_data.get('managed', {}).get('domainStatus', {})
A SSL Certificate resource.
389 @property 390 def full_path(self) -> str: 391 result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', 392 self.self_link) 393 if result: 394 return result.group(1) 395 else: 396 return f'>> {self.self_link}'
Returns the full path of this resource.
Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'
419@caching.cached_api_call(in_memory=True) 420def get_ssl_certificate( 421 project_id: str, 422 certificate_name: str, 423) -> SslCertificate: 424 """Returns object matching certificate name and region""" 425 compute = apis.get_api('compute', 'v1', project_id) 426 427 request = compute.sslCertificates().get(project=project_id, 428 sslCertificate=certificate_name) 429 430 response = request.execute(num_retries=config.API_RETRIES) 431 return SslCertificate(project_id, resource_data=response)
Returns object matching certificate name and region
434class ForwardingRules(models.Resource): 435 """A Forwarding Rule resource.""" 436 437 _resource_data: dict 438 _type: str 439 440 def __init__(self, project_id, resource_data): 441 super().__init__(project_id=project_id) 442 self._resource_data = resource_data 443 444 @property 445 def name(self) -> str: 446 return self._resource_data['name'] 447 448 @property 449 def id(self) -> str: 450 return self._resource_data['id'] 451 452 @property 453 def full_path(self) -> str: 454 result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', 455 self.self_link) 456 if result: 457 return result.group(1) 458 else: 459 return f'>> {self.self_link}' 460 461 @property 462 def short_path(self) -> str: 463 path = self.project_id + '/' + self.name 464 return path 465 466 @property 467 def region(self): 468 url = self._resource_data.get('region', '') 469 if url is not None: 470 match = re.search(r'/([^/]+)/?$', url) 471 if match is not None: 472 region = match.group(1) 473 return region 474 return 'global' 475 476 @property 477 def self_link(self) -> str: 478 return self._resource_data['selfLink'] 479 480 @property 481 def global_access_allowed(self) -> bool: 482 return self._resource_data.get('allowGlobalAccess', False) 483 484 @property 485 def load_balancing_scheme(self) -> str: 486 return self._resource_data.get('loadBalancingScheme', None) 487 488 @property 489 def target(self) -> str: 490 full_path = self._resource_data.get('target', '') 491 result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', full_path) 492 if result: 493 return result.group(1) 494 else: 495 return '' 496 497 @property 498 def backend_service(self) -> str: 499 full_path = self._resource_data.get('backendService', '') 500 result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', full_path) 501 if result: 502 return result.group(1) 503 else: 504 return '' 505 506 @property 507 def ip_address(self) -> str: 508 return self._resource_data.get('IPAddress', '') 509 510 @property 511 def port_range(self) -> str: 512 return self._resource_data.get('portRange', '') 513 514 @caching.cached_api_call(in_memory=True) 515 def get_related_backend_services(self) -> List[BackendServices]: 516 """Returns the backend services related to the forwarding rule.""" 517 if self.backend_service: 518 resource = get_backend_service_by_self_link(self.backend_service) 519 return [resource] if resource else [] 520 if self.target: 521 target_proxy_target = get_target_proxy_reference(self.target) 522 if not target_proxy_target: 523 return [] 524 target_proxy_target_type = target_proxy_target.split('/')[-2] 525 if target_proxy_target_type == 'backendServices': 526 resource = get_backend_service_by_self_link(target_proxy_target) 527 return [resource] if resource else [] 528 elif target_proxy_target_type == 'urlMaps': 529 # Currently it doesn't work for shared-vpc backend services 530 backend_services = get_backend_services(self.project_id) 531 return [ 532 backend_service for backend_service in backend_services 533 if target_proxy_target in backend_service.used_by_refs 534 ] 535 return [] 536 537 @property 538 def load_balancer_type(self) -> LoadBalancerType: 539 target_type = None 540 if self.target: 541 parts = self.target.split('/') 542 if len(parts) >= 2: 543 target_type = parts[-2] 544 545 application_targets = [ 546 'targetHttpProxies', 547 'targetHttpsProxies', 548 'targetGrpcProxies', 549 ] 550 551 return get_load_balancer_type( 552 self.load_balancing_scheme, 553 self.region, 554 'application' if target_type in application_targets else 'network', 555 target_type != 'targetPools', 556 )
A Forwarding Rule resource.
452 @property 453 def full_path(self) -> str: 454 result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', 455 self.self_link) 456 if result: 457 return result.group(1) 458 else: 459 return f'>> {self.self_link}'
Returns the full path of this resource.
Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'
461 @property 462 def short_path(self) -> str: 463 path = self.project_id + '/' + self.name 464 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'
537 @property 538 def load_balancer_type(self) -> LoadBalancerType: 539 target_type = None 540 if self.target: 541 parts = self.target.split('/') 542 if len(parts) >= 2: 543 target_type = parts[-2] 544 545 application_targets = [ 546 'targetHttpProxies', 547 'targetHttpsProxies', 548 'targetGrpcProxies', 549 ] 550 551 return get_load_balancer_type( 552 self.load_balancing_scheme, 553 self.region, 554 'application' if target_type in application_targets else 'network', 555 target_type != 'targetPools', 556 )
559@caching.cached_api_call(in_memory=True) 560def get_target_proxy_reference(target_proxy_self_link: str) -> str: 561 """Retrieves the URL map or backend service associated with a given target proxy. 562 563 Args: 564 target_proxy_self_link: self link of the target proxy 565 566 Returns: 567 The url map or the backend service self link 568 """ 569 target_proxy_type = target_proxy_self_link.split('/')[-2] 570 target_proxy_name = target_proxy_self_link.split('/')[-1] 571 target_proxy_scope = target_proxy_self_link.split('/')[-3] 572 match_result = re.match(r'projects/([^/]+)/', target_proxy_self_link) 573 if not match_result: 574 return '' 575 project_id = match_result.group(1) 576 compute = apis.get_api('compute', 'v1', project_id) 577 578 request = None 579 if target_proxy_type == 'targetHttpsProxies': 580 if target_proxy_scope == 'global': 581 request = compute.targetHttpsProxies().get( 582 project=project_id, targetHttpsProxy=target_proxy_name) 583 else: 584 request = compute.regionTargetHttpsProxies().get( 585 project=project_id, 586 region=target_proxy_scope, 587 targetHttpsProxy=target_proxy_name, 588 ) 589 elif target_proxy_type == 'targetHttpProxies': 590 if target_proxy_scope == 'global': 591 request = compute.targetHttpProxies().get( 592 project=project_id, targetHttpProxy=target_proxy_name) 593 else: 594 request = compute.regionTargetHttpProxies().get( 595 project=project_id, 596 region=target_proxy_scope, 597 targetHttpProxy=target_proxy_name, 598 ) 599 elif target_proxy_type == 'targetTcpProxies': 600 if target_proxy_scope == 'global': 601 request = compute.targetTcpProxies().get(project=project_id, 602 targetTcpProxy=target_proxy_name) 603 else: 604 request = compute.regionTargetTcpProxies().get( 605 project=project_id, 606 region=target_proxy_scope, 607 targetTcpProxy=target_proxy_name, 608 ) 609 elif target_proxy_type == 'targetSslProxies': 610 request = compute.targetSslProxies().get(project=project_id, 611 targetSslProxy=target_proxy_name) 612 elif target_proxy_type == 'targetGrcpProxies': 613 request = compute.targetGrpcProxies().get(project=project_id, 614 targetGrpcProxy=target_proxy_name) 615 if not request: 616 # target is not target proxy 617 return '' 618 response = request.execute(num_retries=config.API_RETRIES) 619 if 'urlMap' in response: 620 return normalize_url(response['urlMap']) 621 if 'service' in response: 622 return normalize_url(response['service']) 623 return ''
Retrieves the URL map or backend service associated with a given target proxy.
Arguments:
- target_proxy_self_link: self link of the target proxy
Returns:
The url map or the backend service self link
626@caching.cached_api_call(in_memory=True) 627def get_forwarding_rules(project_id: str) -> List[ForwardingRules]: 628 logging.debug('fetching Forwarding Rules: %s', project_id) 629 compute = apis.get_api('compute', 'v1', project_id) 630 forwarding_rules = [] 631 request = compute.forwardingRules().aggregatedList(project=project_id) 632 response = request.execute(num_retries=config.API_RETRIES) 633 forwarding_rules_by_region = response['items'] 634 for _, data_ in forwarding_rules_by_region.items(): 635 if 'forwardingRules' not in data_: 636 continue 637 forwarding_rules.extend([ 638 ForwardingRules(project_id, forwarding_rule) 639 for forwarding_rule in data_['forwardingRules'] 640 ]) 641 return forwarding_rules
644@caching.cached_api_call(in_memory=True) 645def get_forwarding_rule(project_id: str, 646 forwarding_rule_name: str, 647 region: str = None) -> ForwardingRules: 648 compute = apis.get_api('compute', 'v1', project_id) 649 if not region or region == 'global': 650 request = compute.globalForwardingRules().get( 651 project=project_id, forwardingRule=forwarding_rule_name) 652 else: 653 request = compute.forwardingRules().get(project=project_id, 654 region=region, 655 forwardingRule=forwarding_rule_name) 656 response = request.execute(num_retries=config.API_RETRIES) 657 return ForwardingRules(project_id, resource_data=response)
660class TargetHttpsProxy(models.Resource): 661 """A Target HTTPS Proxy resource.""" 662 663 _resource_data: dict 664 _type: str 665 666 def __init__(self, project_id, resource_data): 667 super().__init__(project_id=project_id) 668 self._resource_data = resource_data 669 670 @property 671 def name(self) -> str: 672 return self._resource_data['name'] 673 674 @property 675 def id(self) -> str: 676 return self._resource_data['id'] 677 678 @property 679 def full_path(self) -> str: 680 result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', 681 self.self_link) 682 if result: 683 return result.group(1) 684 else: 685 return f'>> {self.self_link}' 686 687 @property 688 def self_link(self) -> str: 689 return self._resource_data['selfLink'] 690 691 @property 692 def region(self): 693 url = self._resource_data.get('region', '') 694 if url is not None: 695 match = re.search(r'/([^/]+)/?$', url) 696 if match is not None: 697 region = match.group(1) 698 return region 699 return 'global' 700 701 @property 702 def ssl_certificates(self) -> List[str]: 703 return self._resource_data.get('sslCertificates', []) 704 705 @property 706 def certificate_map(self) -> str: 707 certificate_map = self._resource_data.get('certificateMap', '') 708 result = re.match(r'https://certificatemanager.googleapis.com/v1/(.*)', 709 certificate_map) 710 if result: 711 return result.group(1) 712 return certificate_map
A Target HTTPS Proxy resource.
678 @property 679 def full_path(self) -> str: 680 result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', 681 self.self_link) 682 if result: 683 return result.group(1) 684 else: 685 return f'>> {self.self_link}'
Returns the full path of this resource.
Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'
715@caching.cached_api_call(in_memory=True) 716def get_target_https_proxies(project_id: str) -> List[TargetHttpsProxy]: 717 logging.debug('fetching Target HTTPS Proxies: %s', project_id) 718 compute = apis.get_api('compute', 'v1', project_id) 719 target_https_proxies = [] 720 request = compute.targetHttpsProxies().aggregatedList(project=project_id) 721 response = request.execute(num_retries=config.API_RETRIES) 722 target_https_proxies_by_region = response['items'] 723 for _, data_ in target_https_proxies_by_region.items(): 724 if 'targetHttpsProxies' not in data_: 725 continue 726 target_https_proxies.extend([ 727 TargetHttpsProxy(project_id, target_https_proxy) 728 for target_https_proxy in data_['targetHttpsProxies'] 729 ]) 730 731 return target_https_proxies
734class TargetSslProxy(models.Resource): 735 """A Target SSL Proxy resource.""" 736 737 _resource_data: dict 738 _type: str 739 740 def __init__(self, project_id, resource_data): 741 super().__init__(project_id=project_id) 742 self._resource_data = resource_data 743 744 @property 745 def name(self) -> str: 746 return self._resource_data['name'] 747 748 @property 749 def id(self) -> str: 750 return self._resource_data['id'] 751 752 @property 753 def full_path(self) -> str: 754 result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', 755 self.self_link) 756 if result: 757 return result.group(1) 758 else: 759 return f'>> {self.self_link}' 760 761 @property 762 def self_link(self) -> str: 763 return self._resource_data['selfLink'] 764 765 @property 766 def region(self): 767 url = self._resource_data.get('region', '') 768 if url is not None: 769 match = re.search(r'/([^/]+)/?$', url) 770 if match is not None: 771 region = match.group(1) 772 return region 773 return 'global' 774 775 @property 776 def ssl_certificates(self) -> List[str]: 777 return self._resource_data.get('sslCertificates', []) 778 779 @property 780 def certificate_map(self) -> str: 781 certificate_map = self._resource_data.get('certificateMap', '') 782 result = re.match(r'https://certificatemanager.googleapis.com/v1/(.*)', 783 certificate_map) 784 if result: 785 return result.group(1) 786 return certificate_map
A Target SSL Proxy resource.
752 @property 753 def full_path(self) -> str: 754 result = re.match(r'https://www.googleapis.com/compute/v1/(.*)', 755 self.self_link) 756 if result: 757 return result.group(1) 758 else: 759 return f'>> {self.self_link}'
Returns the full path of this resource.
Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'
789@caching.cached_api_call(in_memory=True) 790def get_target_ssl_proxies(project_id: str) -> List[TargetSslProxy]: 791 logging.debug('fetching Target SSL Proxies: %s', project_id) 792 compute = apis.get_api('compute', 'v1', project_id) 793 request = compute.targetSslProxies().list(project=project_id) 794 response = request.execute(num_retries=config.API_RETRIES) 795 796 return [ 797 TargetSslProxy(project_id, item) for item in response.get('items', []) 798 ]
801class LoadBalancerInsight(models.Resource): 802 """Represents a Load Balancer Insights object""" 803 804 @property 805 def full_path(self) -> str: 806 return self._resource_data['name'] 807 808 @property 809 def description(self) -> str: 810 return self._resource_data['description'] 811 812 @property 813 def insight_subtype(self) -> str: 814 return self._resource_data['insightSubtype'] 815 816 @property 817 def details(self) -> dict: 818 return self._resource_data['content'] 819 820 @property 821 def is_firewall_rule_insight(self) -> bool: 822 firewall_rule_subtypes = ( 823 'HEALTH_CHECK_FIREWALL_NOT_CONFIGURED', 824 'HEALTH_CHECK_FIREWALL_FULLY_BLOCKING', 825 'HEALTH_CHECK_FIREWALL_PARTIALLY_BLOCKING', 826 'HEALTH_CHECK_FIREWALL_INCONSISTENT', 827 ) 828 return self.insight_subtype.startswith(firewall_rule_subtypes) 829 830 @property 831 def is_health_check_port_mismatch_insight(self) -> bool: 832 return self.insight_subtype == 'HEALTH_CHECK_PORT_MISMATCH' 833 834 def __init__(self, project_id, resource_data): 835 super().__init__(project_id=project_id) 836 self._resource_data = resource_data
Represents a Load Balancer Insights object
Returns the full path of this resource.
Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'
820 @property 821 def is_firewall_rule_insight(self) -> bool: 822 firewall_rule_subtypes = ( 823 'HEALTH_CHECK_FIREWALL_NOT_CONFIGURED', 824 'HEALTH_CHECK_FIREWALL_FULLY_BLOCKING', 825 'HEALTH_CHECK_FIREWALL_PARTIALLY_BLOCKING', 826 'HEALTH_CHECK_FIREWALL_INCONSISTENT', 827 ) 828 return self.insight_subtype.startswith(firewall_rule_subtypes)
839@caching.cached_api_call 840def get_lb_insights_for_a_project(project_id: str, region: str = 'global'): 841 api = apis.get_api('recommender', 'v1', project_id) 842 843 insight_name = (f'projects/{project_id}/locations/{region}/insightTypes/' 844 'google.networkanalyzer.networkservices.loadBalancerInsight') 845 insights = [] 846 for insight in apis_utils.list_all( 847 request=api.projects().locations().insightTypes().insights().list( 848 parent=insight_name), 849 next_function=api.projects().locations().insightTypes().insights(). 850 list_next, 851 response_keyword='insights', 852 ): 853 insights.append(LoadBalancerInsight(project_id, insight)) 854 return insights