gcpdiag.queries.lb

25class BackendServices(models.Resource):
26  """A Backend Service resource."""
27
28  _resource_data: dict
29  _type: str
30
31  def __init__(self, project_id, resource_data):
32    super().__init__(project_id=project_id)
33    self._resource_data = resource_data
34
35  @property
36  def name(self) -> str:
37    return self._resource_data['name']
38
39  @property
40  def id(self) -> str:
41    return self._resource_data['id']
42
43  @property
44  def full_path(self) -> str:
45    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)',
46                      self.self_link)
47    if result:
48      return result.group(1)
49    else:
50      return f'>> {self.self_link}'
51
52  @property
53  def short_path(self) -> str:
54    path = self.project_id + '/' + self.name
55    return path
56
57  @property
58  def self_link(self) -> str:
59    return self._resource_data['selfLink']
60
61  @property
62  def session_affinity(self) -> str:
63    return self._resource_data.get('sessionAffinity', 'NONE')
64
65  @property
66  def locality_lb_policy(self) -> str:
67    return self._resource_data.get('localityLbPolicy', 'ROUND_ROBIN')
68
69  @property
70  def is_enable_cdn(self) -> str:
71    return self._resource_data.get('enableCDN', False)
72
73  @property
74  def load_balancing_scheme(self) -> str:
75    return self._resource_data.get('loadBalancingScheme', 'NONE')
76
77  @property
78  def health_check(self) -> str:
79    health_check_url = self._resource_data['healthChecks'][0]
80    matches = re.search(r'/([^/]+)$', health_check_url)
81    if matches:
82      healthcheck_name = matches.group(1)
83      return healthcheck_name
84    else:
85      return ''
86
87  @property
88  def region(self):
89    try:
90      url = self._resource_data.get('region', 'NONE')
91      if url is not None:
92        match = re.search(r'/([^/]+)/?$', url)
93        if match is not None:
94          region = match.group(1)
95          return region
96        else:
97          return 'None'
98    except KeyError:
99      return 'None'

102@caching.cached_api_call(in_memory=True)
103def get_backend_services(project_id: str) -> List[BackendServices]:
104  logging.info('fetching Backend Services: %s', project_id)
105  compute = apis.get_api('compute', 'v1', project_id)
106  request = compute.backendServices().list(project=project_id)
107  response = request.execute(num_retries=config.API_RETRIES)
108  return [
109      BackendServices(project_id, item) for item in response.get('items', [])
110  ]

113class ForwardingRules(models.Resource):
114  """A Forwarding Rule resource."""
115  _resource_data: dict
116  _type: str
117
118  def __init__(self, project_id, resource_data):
119    super().__init__(project_id=project_id)
120    self._resource_data = resource_data
121
122  @property
123  def name(self) -> str:
124    return self._resource_data['name']
125
126  @property
127  def id(self) -> str:
128    return self._resource_data['id']
129
130  @property
131  def full_path(self) -> str:
132    result = re.match(r'https://www.googleapis.com/compute/v1/(.*)',
133                      self.self_link)
134    if result:
135      return result.group(1)
136    else:
137      return f'>> {self.self_link}'
138
139  @property
140  def short_path(self) -> str:
141    path = self.project_id + '/' + self.name
142    return path
143
144  @property
145  def self_link(self) -> str:
146    return self._resource_data['selfLink']
147
148  @property
149  def is_global_access_allowed(self) -> bool:
150    return self._resource_data.get('allowGlobalAccess', False)

153@caching.cached_api_call(in_memory=True)
154def get_forwarding_rules(project_id: str) -> List[ForwardingRules]:
155  logging.info('fetching Forwarding Rules: %s', project_id)
156  compute = apis.get_api('compute', 'v1', project_id)
157  forwarding_rules = []
158  request = compute.forwardingRules().aggregatedList(project=project_id)
159  response = request.execute(num_retries=config.API_RETRIES)
160  forwarding_rules_by_region = response['items']
161  for _, data_ in forwarding_rules_by_region.items():
162    if 'forwardingRules' not in data_:
163      continue
164    forwarding_rules.extend([
165        ForwardingRules(project_id, forwarding_rule)
166        for forwarding_rule in data_['forwardingRules']
167    ])
168  return forwarding_rules