gcpdiag.queries.datafusion

 37class Instance(models.Resource):
 38  """Represents a Data Fusion instance.
 39
 40  https://cloud.google.com/data-fusion/docs/reference/rest/v1/projects.locations.instances#Instance
 41  """
 42
 43  _resource_data: dict
 44
 45  def __init__(self, project_id, resource_data):
 46    super().__init__(project_id=project_id)
 47    self._resource_data = resource_data
 48
 49  @property
 50  def full_path(self) -> str:
 51    """
 52    The 'name' of the instance is already in the full path form
 53
 54    projects/{project}/locations/{location}/instances/{instance}.
 55    """
 56    return self._resource_data['name']
 57
 58  @property
 59  def short_path(self) -> str:
 60    path = self.full_path
 61    path = re.sub(r'^projects/', '', path)
 62    path = re.sub(r'/locations/', '/', path)
 63    path = re.sub(r'/instances/', '/', path)
 64    return path
 65
 66  @property
 67  def name(self) -> str:
 68    return utils.extract_value_from_res_name(self._resource_data['name'], 'instances')
 69
 70  @property
 71  def location(self) -> str:
 72    return utils.extract_value_from_res_name(self._resource_data['name'], 'locations')
 73
 74  @property
 75  def zone(self) -> str:
 76    return self._resource_data['zone']
 77
 78  @property
 79  def type(self) -> str:
 80    return self._resource_data['type']
 81
 82  @property
 83  def is_basic_type(self) -> bool:
 84    return self._resource_data['type'] == 'BASIC'
 85
 86  @property
 87  def is_enterprise_type(self) -> bool:
 88    return self._resource_data['type'] == 'ENTERPRISE'
 89
 90  @property
 91  def is_developer_type(self) -> bool:
 92    return self._resource_data['type'] == 'DEVELOPER'
 93
 94  @property
 95  def is_private(self) -> bool:
 96    if 'privateInstance' in self._resource_data:
 97      return self._resource_data['privateInstance']
 98    return False
 99
100  @property
101  def status(self) -> str:
102    return self._resource_data['state']
103
104  @property
105  def status_details(self) -> Optional[str]:
106    if 'stateMessage' in self._resource_data:
107      return self._resource_data['stateMessage']
108    return None
109
110  @property
111  def is_running(self) -> bool:
112    return self.status == 'ACTIVE'
113
114  @property
115  def is_deleting(self) -> bool:
116    return self._resource_data['state'] == 'DELETING'
117
118  @property
119  def version(self) -> Version:
120    return Version(self._resource_data['version'])
121
122  @property
123  def api_service_agent(self) -> str:
124    return self._resource_data['p4ServiceAccount']
125
126  @property
127  def dataproc_service_account(self) -> str:
128    sa = self._resource_data.get('dataprocServiceAccount')
129    if sa is None:
130      sa = crm.get_project(self.project_id).default_compute_service_account
131    return sa
132
133  @property
134  def tenant_project_id(self) -> str:
135    return self._resource_data['tenantProjectId']
136
137  @property
138  def uses_shared_vpc(self) -> bool:
139    """
140    If shared VPC then 'network_string' = 'projects/{host-project-id}/global/networks/{network}'
141    else 'network_string' = {network}
142    """
143    if 'network' in self._resource_data['networkConfig']:
144      network_string = self._resource_data['networkConfig']['network']
145      match = re.match(r'projects/([^/]+)/global/networks/([^/]+)$', network_string)
146      if match and match.group(1) != self.project_id:
147        return True
148
149    return False
150
151  @property
152  def network(self) -> network.Network:
153    if 'network' in self._resource_data['networkConfig']:
154      network_string = self._resource_data['networkConfig']['network']
155      match = re.match(r'projects/([^/]+)/global/networks/([^/]+)$', network_string)
156      if match:
157        return network.get_network(
158          match.group(1),
159          match.group(2),
160          context=models.Context(project_id=match.group(1)),
161        )
162      else:
163        return network.get_network(
164          self.project_id,
165          network_string,
166          context=models.Context(project_id=self.project_id),
167        )
168
169    return network.get_network(
170      self.project_id,
171      'default',
172      context=models.Context(project_id=self.project_id),
173    )
174
175  @property
176  def tp_ipv4_cidr(self) -> Optional[IPv4NetOrIPv6Net]:
177    if 'network' in self._resource_data['networkConfig']:
178      cidr = self._resource_data['networkConfig']['ipAllocation']
179      return ipaddress.ip_network(cidr)
180    return None
181
182  @property
183  def api_endpoint(self) -> str:
184    return self._resource_data['apiEndpoint']

187@caching.cached_api_call
188def get_instances(context: models.Context) -> Mapping[str, Instance]:
189  """Get a dict of Instance matching the given context, indexed by instance full path."""
190  instances: Dict[str, Instance] = {}
191
192  if not apis.is_enabled(context.project_id, 'datafusion'):
193    return instances
194
195  logging.debug('fetching list of Data Fusion instances in project %s', context.project_id)
196  datafusion_api = apis.get_api('datafusion', 'v1', context.project_id)
197  query = (
198    datafusion_api.projects()
199    .locations()
200    .instances()
201    .list(parent=f'projects/{context.project_id}/locations/-')
202  )  #'-' (wildcard) all regions
203
204  try:
205    resp = query.execute(num_retries=config.API_RETRIES)
206    if 'instances' not in resp:
207      return instances
208
209    for i in resp['instances']:
210      # projects/{project}/locations/{location}/instances/{instance}.
211      result = re.match(r'projects/[^/]+/locations/([^/]+)/instances/([^/]+)', i['name'])
212      if not result:
213        logging.error('invalid datafusion name: %s', i['name'])
214        continue
215      location = result.group(1)
216      labels = i.get('labels', {})
217      name = result.group(2)
218      if not context.match_project_resource(location=location, labels=labels, resource=name):
219        continue
220
221      instances[i['name']] = Instance(project_id=context.project_id, resource_data=i)
222
223  except googleapiclient.errors.HttpError as err:
224    raise utils.GcpApiError(err) from err
225
226  return instances

229@caching.cached_api_call
230def extract_support_datafusion_version() -> Dict[str, str]:
231  """Extract the version policy dictionary from the data fusion version support policy page.
232
233  Returns:
234    A dictionary of data fusion versions and their support end dates.
235  """
236  page_url = 'https://cloud.google.com/data-fusion/docs/support/version-support-policy'
237
238  try:
239    data_fusion_table = web.fetch_and_extract_table(page_url, tag='h2', tag_id='support_timelines')
240    if data_fusion_table:
241      versions = []
242      support_end_dates = []
243      version_policy_dict = {}
244
245      for row in data_fusion_table.find_all('tr')[1:]:
246        columns = row.find_all('td')
247        version = columns[0]
248        support_end_date = columns[2].text.strip()
249        if version.sup:
250          version.sup.decompose()
251
252        version = version.text.strip()
253        try:
254          support_end_date = datetime.datetime.strptime(support_end_date, '%B %d, %Y')
255          support_end_date = datetime.datetime.strftime(support_end_date, '%Y-%m-%d')
256        except ValueError:
257          continue
258
259        versions.append(version)
260        support_end_dates.append(support_end_date)
261
262        version_policy_dict = dict(zip(versions, support_end_dates))
263      return version_policy_dict
264
265    else:
266      return {}
267
268  except (
269    requests.exceptions.RequestException,
270    AttributeError,
271    TypeError,
272    ValueError,
273    IndexError,
274  ) as e:
275    logging.error('Error in extracting data fusion version support policy: %s', e)
276    return {}

279class Profile(models.Resource):
280  """Represents a Compute Profile."""
281
282  _resource_data: dict
283
284  def __init__(self, project_id, instance_name, resource_data):
285    super().__init__(project_id=project_id)
286    self.instance_name = instance_name
287    self._resource_data = resource_data
288
289  @property
290  def full_path(self) -> str:
291    """The full path form :
292
293    projects/{project}/instances/{instance}/computeProfiles/{profile}.
294    """
295    return (
296      f'projects/{self.project_id}/instances/{self.instance_name}'
297      f'/computeProfiles/{self._resource_data["name"]}'
298    )
299
300  @property
301  def short_path(self) -> str:
302    """The short path form :
303
304    {project}/{instance}/{profile}.
305    """
306    return f'{self.project_id}/{self.instance_name}/{self._resource_data["name"]}'
307
308  @property
309  def name(self) -> str:
310    return self._resource_data['name']
311
312  @property
313  def region(self) -> str:
314    for value in self._resource_data['provisioner'].get('properties'):
315      if value.get('name') == 'region' and value.get('value') is not None:
316        return value.get('value')
317    return 'No region defined'
318
319  @property
320  def status(self) -> str:
321    return self._resource_data['status']
322
323  @property
324  def scope(self) -> str:
325    return self._resource_data['scope']
326
327  @property
328  def is_dataproc_provisioner(self) -> bool:
329    return self._resource_data['provisioner']['name'] == 'gcp-dataproc'
330
331  @property
332  def is_existing_dataproc_provisioner(self) -> bool:
333    return self._resource_data['provisioner']['name'] == 'gcp-existing-dataproc'
334
335  @property
336  def autoscaling_enabled(self) -> bool:
337    for value in self._resource_data['provisioner'].get('properties'):
338      if value.get('name') == 'enablePredefinedAutoScaling' and value.get('value') is not None:
339        return value.get('value') == 'true'
340    return False
341
342  @property
343  def image_version(self) -> str:
344    for value in self._resource_data['provisioner'].get('properties'):
345      if value.get('name') == 'imageVersion' and value.get('value') != '':
346        return value.get('value')
347    return 'No imageVersion defined'
348
349  @property
350  def auto_scaling_policy(self) -> str:
351    for value in self._resource_data['provisioner'].get('properties'):
352      if value.get('name') == 'autoScalingPolicy' and value.get('value') != '':
353        return value.get('value')
354    return 'No autoScalingPolicy defined'

357@caching.cached_api_call
358def get_instance_system_compute_profile(
359  context: models.Context, instance: Instance
360) -> Iterable[Profile]:
361  """Get a list of datafusion Instance dataproc System compute profile."""
362  logging.debug('fetching dataproc System compute profile list: %s', context.project_id)
363  system_profiles: List[Profile] = []
364  cdap_endpoint = instance.api_endpoint
365  datafusion = get_generic.get_generic_api('datafusion', cdap_endpoint)
366  response = datafusion.get_system_profiles()
367  if response is not None:
368    for res in response:
369      if (
370        res['provisioner']['name'] == 'gcp-dataproc'
371        or res['provisioner']['name'] == 'gcp-existing-dataproc'
372      ):
373        system_profiles.append(Profile(context.project_id, instance.name, res))
374  return system_profiles

377@caching.cached_api_call
378def get_instance_user_compute_profile(
379  context: models.Context, instance: Instance
380) -> Iterable[Profile]:
381  """Get a list of datafusion Instance dataproc User compute profile."""
382  logging.debug('fetching dataproc User compute profile list: %s', context.project_id)
383  user_profiles: List[Profile] = []
384  cdap_endpoint = instance.api_endpoint
385  datafusion = get_generic.get_generic_api('datafusion', cdap_endpoint)
386  response_namespaces = datafusion.get_all_namespaces()
387  if response_namespaces is not None:
388    for res in response_namespaces:
389      response = datafusion.get_user_profiles(namespace=res['name'])
390      if response is not None:
391        for res in response:
392          if (
393            res['provisioner']['name'] == 'gcp-dataproc'
394            or res['provisioner']['name'] == 'gcp-existing-dataproc'
395          ):
396            user_profiles.append(Profile(context.project_id, instance.name, res))
397      user_profiles = list(filter(bool, user_profiles))
398  return user_profiles

401@caching.cached_api_call
402def extract_datafusion_dataproc_version() -> Dict[str, list[str]]:
403  """Extract the supported Data Fusion versions and their corresponding
404  Dataproc versions from the GCP documentation."""
405
406  page_url = 'https://cloud.google.com/data-fusion/docs/concepts/configure-clusters'
407
408  try:
409    table = web.fetch_and_extract_table(page_url, tag='h2', tag_id='version-compatibility')
410    if table:
411      rows = table.find_all('tr')[1:]  # Skip the header row
412      version_dict = {}
413
414      for row in rows:
415        cdf_versions = row.find_all('td')[0].get_text().strip()
416        dp_versions = row.find_all('td')[1].get_text().strip()
417
418        cdf_versions = cdf_versions.replace(' and later', '')
419        cdf_versions_list = []
420
421        if '-' in cdf_versions:
422          start, end = map(float, cdf_versions.split('-'))
423          while start <= end:
424            cdf_versions_list.append(f'{start:.1f}')
425            start += 0.1
426        else:
427          cdf_versions_list.append(cdf_versions)
428        dp_versions = [v.split('*')[0].strip() for v in dp_versions.split(',')]
429        for version in cdf_versions_list:
430          version_dict[version] = dp_versions
431      return version_dict
432
433    else:
434      return {}
435  except (
436    requests.exceptions.RequestException,
437    AttributeError,
438    TypeError,
439    ValueError,
440    IndexError,
441  ) as e:
442    logging.error(
443      'Error in extracting datafusion and dataproc versions: %s',
444      e,
445    )
446    return {}

449class Preference(models.Resource):
450  """Represents a Preference."""
451
452  _resource_data: dict
453
454  def __init__(self, project_id, instance, resource_data):
455    super().__init__(project_id=project_id)
456    self.instance = instance
457    self._resource_data = resource_data
458
459  @property
460  def full_path(self) -> str:
461    """The full path form :
462
463    projects/{project}/locations/{location}/instances/{instance}.
464    """
465    return self.instance.full_path
466
467  @property
468  def image_version(self):
469    return self._resource_data.get('system.profile.properties.imageVersion', None)

472def get_system_preferences(context: models.Context, instance: Instance) -> Preference:
473  """Get datafusion Instance system preferences."""
474  logging.debug('fetching dataproc System preferences: %s', context.project_id)
475  cdap_endpoint = instance.api_endpoint
476  datafusion = get_generic.get_generic_api('datafusion', cdap_endpoint)
477  response = datafusion.get_system_preferences()
478  return Preference(context.project_id, instance, response)

481def get_namespace_preferences(
482  context: models.Context, instance: Instance
483) -> Mapping[str, Preference]:
484  """Get datafusion cdap namespace preferences."""
485  logging.debug('fetching dataproc namespace preferences: %s', context.project_id)
486  cdap_endpoint = instance.api_endpoint
487  datafusion = get_generic.get_generic_api('datafusion', cdap_endpoint)
488  namespaces = datafusion.get_all_namespaces()
489  namespaces_preferences = {}
490  if namespaces is not None:
491    for namespace in namespaces:
492      response = datafusion.get_namespace_preferences(namespace=namespace['name'])
493      if bool(response):
494        namespaces_preferences[namespace['name']] = Preference(
495          context.project_id, instance, response
496        )
497  return namespaces_preferences

500def get_application_preferences(
501  context: models.Context, instance: Instance
502) -> Mapping[str, Preference]:
503  """Get datafusion cdap application preferences."""
504  logging.debug('fetching dataproc application preferences: %s', context.project_id)
505  cdap_endpoint = instance.api_endpoint
506  datafusion = get_generic.get_generic_api('datafusion', cdap_endpoint)
507  applications_preferences = {}
508  namespaces = datafusion.get_all_namespaces()
509  if namespaces is not None:
510    for namespace in namespaces:
511      applications = datafusion.get_all_applications(namespace=namespace['name'])
512      if applications is not None:
513        for application in applications:
514          response = datafusion.get_application_preferences(
515            namespace=namespace['name'], application_name=application['name']
516          )
517          if bool(response):
518            applications_preferences[application['name']] = Preference(
519              context.project_id, instance, response
520            )
521  return applications_preferences