gcpdiag.queries.lb

Queries related to load balancer.
class LoadBalancerType(enum.Enum):
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.

LOAD_BALANCER_TYPE_UNSPECIFIED = <LoadBalancerType.LOAD_BALANCER_TYPE_UNSPECIFIED: 0>
EXTERNAL_PASSTHROUGH_LB = <LoadBalancerType.EXTERNAL_PASSTHROUGH_LB: 1>
INTERNAL_PASSTHROUGH_LB = <LoadBalancerType.INTERNAL_PASSTHROUGH_LB: 2>
TARGET_POOL_LB = <LoadBalancerType.TARGET_POOL_LB: 3>
GLOBAL_EXTERNAL_PROXY_NETWORK_LB = <LoadBalancerType.GLOBAL_EXTERNAL_PROXY_NETWORK_LB: 4>
REGIONAL_INTERNAL_PROXY_NETWORK_LB = <LoadBalancerType.REGIONAL_INTERNAL_PROXY_NETWORK_LB: 5>
REGIONAL_EXTERNAL_PROXY_NETWORK_LB = <LoadBalancerType.REGIONAL_EXTERNAL_PROXY_NETWORK_LB: 6>
CROSS_REGION_INTERNAL_PROXY_NETWORK_LB = <LoadBalancerType.CROSS_REGION_INTERNAL_PROXY_NETWORK_LB: 7>
CLASSIC_PROXY_NETWORK_LB = <LoadBalancerType.CLASSIC_PROXY_NETWORK_LB: 8>
GLOBAL_EXTERNAL_APPLICATION_LB = <LoadBalancerType.GLOBAL_EXTERNAL_APPLICATION_LB: 9>
REGIONAL_INTERNAL_APPLICATION_LB = <LoadBalancerType.REGIONAL_INTERNAL_APPLICATION_LB: 10>
REGIONAL_EXTERNAL_APPLICATION_LB = <LoadBalancerType.REGIONAL_EXTERNAL_APPLICATION_LB: 11>
CROSS_REGION_INTERNAL_APPLICATION_LB = <LoadBalancerType.CROSS_REGION_INTERNAL_APPLICATION_LB: 12>
CLASSIC_APPLICATION_LB = <LoadBalancerType.CLASSIC_APPLICATION_LB: 13>
def get_load_balancer_type_name(lb_type: LoadBalancerType) -> str:
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.

def get_load_balancer_type( load_balancing_scheme: str, scope: str, layer: Literal['application', 'network'], backend_service_based: bool = True) -> LoadBalancerType:
 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
def normalize_url(url: str) -> str:
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.

class BackendServices(gcpdiag.models.Resource):
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.

BackendServices(project_id, resource_data)
140  def __init__(self, project_id, resource_data):
141    super().__init__(project_id=project_id)
142    self._resource_data = resource_data
name: str
144  @property
145  def name(self) -> str:
146    return self._resource_data['name']
id: str
148  @property
149  def id(self) -> str:
150    return self._resource_data['id']
full_path: str
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'

short_path: str
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'

session_affinity: str
170  @property
171  def session_affinity(self) -> str:
172    return self._resource_data.get('sessionAffinity', 'NONE')
timeout_sec: int
174  @property
175  def timeout_sec(self) -> int:
176    return self._resource_data.get('timeoutSec', None)
locality_lb_policy: str
178  @property
179  def locality_lb_policy(self) -> str:
180    return self._resource_data.get('localityLbPolicy', 'ROUND_ROBIN')
is_enable_cdn: str
182  @property
183  def is_enable_cdn(self) -> str:
184    return self._resource_data.get('enableCDN', False)
draining_timeout_sec: int
186  @property
187  def draining_timeout_sec(self) -> int:
188    return self._resource_data.get('connectionDraining',
189                                   {}).get('drainingTimeoutSec', 0)
load_balancing_scheme: str
191  @property
192  def load_balancing_scheme(self) -> str:
193    return self._resource_data.get('loadBalancingScheme', None)
health_check: str
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 ''
backends: List[dict]
205  @property
206  def backends(self) -> List[dict]:
207    return self._resource_data.get('backends', [])
region
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
protocol: str
223  @property
224  def protocol(self) -> str:
225    return self._resource_data.get('protocol', None)
used_by_refs: List[str]
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
load_balancer_type: LoadBalancerType
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    )
@caching.cached_api_call(in_memory=True)
def get_backend_services(project_id: str) -> List[BackendServices]:
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
@caching.cached_api_call(in_memory=True)
def get_backend_service( project_id: str, backend_service_name: str, region: str = None) -> BackendServices:
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

class BackendHealth:
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.

BackendHealth(resource_data, group)
307  def __init__(self, resource_data, group):
308    self._resource_data = resource_data
309    self._group = group
instance: str
311  @property
312  def instance(self) -> str:
313    return self._resource_data['instance']
group: str
315  @property
316  def group(self) -> str:
317    return self._group
health_state: str
319  @property
320  def health_state(self) -> str:
321    return self._resource_data.get('healthState', 'UNHEALTHY')
@caching.cached_api_call(in_memory=True)
def get_backend_service_health( project_id: str, backend_service_name: str, backend_service_region: str = None) -> List[BackendHealth]:
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.

class SslCertificate(gcpdiag.models.Resource):
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.

SslCertificate(project_id, resource_data)
377  def __init__(self, project_id, resource_data):
378    super().__init__(project_id=project_id)
379    self._resource_data = resource_data
name: str
381  @property
382  def name(self) -> str:
383    return self._resource_data['name']
id: str
385  @property
386  def id(self) -> str:
387    return self._resource_data['id']
full_path: str
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'

type: str
402  @property
403  def type(self) -> str:
404    return self._resource_data.get('type', 'SELF_MANAGED')
status: str
406  @property
407  def status(self) -> str:
408    return self._resource_data.get('managed', {}).get('status')
domains: List[str]
410  @property
411  def domains(self) -> List[str]:
412    return self._resource_data.get('managed', {}).get('domains', [])
domain_status: Dict[str, str]
414  @property
415  def domain_status(self) -> Dict[str, str]:
416    return self._resource_data.get('managed', {}).get('domainStatus', {})
@caching.cached_api_call(in_memory=True)
def get_ssl_certificate( project_id: str, certificate_name: str) -> SslCertificate:
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

class ForwardingRules(gcpdiag.models.Resource):
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.

ForwardingRules(project_id, resource_data)
440  def __init__(self, project_id, resource_data):
441    super().__init__(project_id=project_id)
442    self._resource_data = resource_data
name: str
444  @property
445  def name(self) -> str:
446    return self._resource_data['name']
id: str
448  @property
449  def id(self) -> str:
450    return self._resource_data['id']
full_path: str
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'

short_path: str
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'

region
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'
global_access_allowed: bool
480  @property
481  def global_access_allowed(self) -> bool:
482    return self._resource_data.get('allowGlobalAccess', False)
load_balancing_scheme: str
484  @property
485  def load_balancing_scheme(self) -> str:
486    return self._resource_data.get('loadBalancingScheme', None)
target: str
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 ''
backend_service: str
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 ''
ip_address: str
506  @property
507  def ip_address(self) -> str:
508    return self._resource_data.get('IPAddress', '')
port_range: str
510  @property
511  def port_range(self) -> str:
512    return self._resource_data.get('portRange', '')
load_balancer_type: LoadBalancerType
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    )
@caching.cached_api_call(in_memory=True)
def get_target_proxy_reference(target_proxy_self_link: str) -> str:
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

@caching.cached_api_call(in_memory=True)
def get_forwarding_rules(project_id: str) -> List[ForwardingRules]:
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
@caching.cached_api_call(in_memory=True)
def get_forwarding_rule( project_id: str, forwarding_rule_name: str, region: str = None) -> ForwardingRules:
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)
class TargetHttpsProxy(gcpdiag.models.Resource):
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.

TargetHttpsProxy(project_id, resource_data)
666  def __init__(self, project_id, resource_data):
667    super().__init__(project_id=project_id)
668    self._resource_data = resource_data
name: str
670  @property
671  def name(self) -> str:
672    return self._resource_data['name']
id: str
674  @property
675  def id(self) -> str:
676    return self._resource_data['id']
full_path: str
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'

region
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'
ssl_certificates: List[str]
701  @property
702  def ssl_certificates(self) -> List[str]:
703    return self._resource_data.get('sslCertificates', [])
certificate_map: str
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
@caching.cached_api_call(in_memory=True)
def get_target_https_proxies(project_id: str) -> List[TargetHttpsProxy]:
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
class TargetSslProxy(gcpdiag.models.Resource):
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.

TargetSslProxy(project_id, resource_data)
740  def __init__(self, project_id, resource_data):
741    super().__init__(project_id=project_id)
742    self._resource_data = resource_data
name: str
744  @property
745  def name(self) -> str:
746    return self._resource_data['name']
id: str
748  @property
749  def id(self) -> str:
750    return self._resource_data['id']
full_path: str
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'

region
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'
ssl_certificates: List[str]
775  @property
776  def ssl_certificates(self) -> List[str]:
777    return self._resource_data.get('sslCertificates', [])
certificate_map: str
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
@caching.cached_api_call(in_memory=True)
def get_target_ssl_proxies(project_id: str) -> List[TargetSslProxy]:
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  ]
class LoadBalancerInsight(gcpdiag.models.Resource):
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

LoadBalancerInsight(project_id, resource_data)
834  def __init__(self, project_id, resource_data):
835    super().__init__(project_id=project_id)
836    self._resource_data = resource_data
full_path: str
804  @property
805  def full_path(self) -> str:
806    return self._resource_data['name']

Returns the full path of this resource.

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

description: str
808  @property
809  def description(self) -> str:
810    return self._resource_data['description']
insight_subtype: str
812  @property
813  def insight_subtype(self) -> str:
814    return self._resource_data['insightSubtype']
details: dict
816  @property
817  def details(self) -> dict:
818    return self._resource_data['content']
is_firewall_rule_insight: bool
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)
is_health_check_port_mismatch_insight: bool
830  @property
831  def is_health_check_port_mismatch_insight(self) -> bool:
832    return self.insight_subtype == 'HEALTH_CHECK_PORT_MISMATCH'
@caching.cached_api_call
def get_lb_insights_for_a_project(project_id: str, region: str = 'global'):
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