gcpdiag.queries.dataproc

 29class Cluster(models.Resource):
 30  """Represents Dataproc Cluster"""
 31
 32  name: str
 33  _resource_data: Mapping
 34
 35  def __init__(self, name: str, project_id: str, resource_data: Mapping):
 36    super().__init__(project_id)
 37    self.name = name
 38    self._resource_data = resource_data
 39
 40  def is_running(self) -> bool:
 41    return self.status == 'RUNNING'
 42
 43  def get_software_property(self, property_name) -> str:
 44    return self._resource_data['config']['softwareConfig']['properties'].get(property_name)
 45
 46  def is_stackdriver_logging_enabled(self) -> bool:
 47    # Unless overridden during create,
 48    # properties with default values are not returned,
 49    # therefore get_software_property should only return when its false
 50    return not self.get_software_property('dataproc:dataproc.logging.stackdriver.enable') == 'false'
 51
 52  def is_stackdriver_monitoring_enabled(self) -> bool:
 53    return self.get_software_property('dataproc:dataproc.monitoring.stackdriver.enable') == 'true'
 54
 55  @property
 56  def region(self) -> str:
 57    """biggest regions have a trailing '-d' at most in its zoneUri
 58
 59    https://www.googleapis.com/compute/v1/projects/dataproc1/zones/us-central1-d
 60    """
 61    return self._resource_data['config']['gceClusterConfig']['zoneUri'].split('/')[-1][0:-2]
 62
 63  @property
 64  def zone(self) -> Optional[str]:
 65    zone = self._resource_data.get('config', {}).get('gceClusterConfig', {}).get('zoneUri')
 66    if zone:
 67      m = re.search(r'/zones/([^/]+)$', zone)
 68      if m:
 69        return m.group(1)
 70    raise RuntimeError(f"can't determine zone for cluster {self.name}")
 71
 72  @property
 73  def full_path(self) -> str:
 74    return f'projects/{self.project_id}/regions/{self.region}/clusters/{self.name}'
 75
 76  @property
 77  def short_path(self) -> str:
 78    return f'{self.project_id}/{self.region}/{self.name}'
 79
 80  @property
 81  def status(self) -> str:
 82    return self._resource_data['status']['state']
 83
 84  def __str__(self) -> str:
 85    return self.short_path
 86
 87  @property
 88  def cluster_uuid(self) -> str:
 89    return self._resource_data['clusterUuid']
 90
 91  @property
 92  def image_version(self):
 93    return self._resource_data['config']['softwareConfig']['imageVersion']
 94
 95  @property
 96  def vm_service_account_email(self):
 97    sa = self._resource_data['config']['gceClusterConfig'].get('serviceAccount')
 98    if sa is None:
 99      sa = crm.get_project(self.project_id).default_compute_service_account
100    return sa
101
102  @property
103  def is_custom_gcs_connector(self) -> bool:
104    return bool(
105      self._resource_data.get('config', {})
106      .get('gceClusterConfig', {})
107      .get('metadata', {})
108      .get('GCS_CONNECTOR_VERSION')
109    )
110
111  @property
112  def cluster_provided_bq_connector(self):
113    """Check user-supplied BigQuery connector on the cluster level"""
114    bigquery_connector = (
115      self._resource_data.get('config', {})
116      .get('gceClusterConfig', {})
117      .get('metadata', {})
118      .get('SPARK_BQ_CONNECTOR_VERSION')
119    )
120    if not bigquery_connector:
121      bigquery_connector = (
122        self._resource_data.get('config', {})
123        .get('gceClusterConfig', {})
124        .get('metadata', {})
125        .get('SPARK_BQ_CONNECTOR_URL')
126      )
127      if bigquery_connector:
128        if bigquery_connector == 'spark-bigquery-latest.jar':
129          return 'spark-bigquery-latest'
130        else:
131          match = re.search(
132            r'spark-bigquery(?:-with-dependencies_\d+\.\d+)?-(\d+\.\d+\.\d+)\.jar',
133            bigquery_connector,
134          )
135          if match:
136            return match.group(1)
137      # If returns None, it means that the cluster is using the default,
138      # pre-installed BQ connector for the image version
139      return bigquery_connector
140
141  @property
142  def is_gce_cluster(self) -> bool:
143    return bool(self._resource_data.get('config', {}).get('gceClusterConfig'))
144
145  @property
146  def gce_network_uri(self) -> Optional[str]:
147    """Get network uri from cluster network or subnetwork"""
148    if not self.is_gce_cluster:
149      raise RuntimeError('Can not return network URI for a Dataproc on GKE cluster')
150    network_uri = (
151      self._resource_data.get('config', {}).get('gceClusterConfig', {}).get('networkUri')
152    )
153    if not network_uri:
154      subnetwork_uri = (
155        self._resource_data.get('config', {}).get('gceClusterConfig', {}).get('subnetworkUri')
156      )
157      network_uri = network.get_subnetwork_from_url(subnetwork_uri).network
158    return network_uri
159
160  @property
161  def gce_subnetwork_uri(self) -> Optional[str]:
162    """Get subnetwork uri from cluster subnetwork."""
163    if not self.is_gce_cluster:
164      raise RuntimeError('Can not return subnetwork URI for a Dataproc on GKE cluster')
165    subnetwork_uri = (
166      self._resource_data.get('config', {}).get('gceClusterConfig', {}).get('subnetworkUri')
167    )
168    if not subnetwork_uri:
169      subnetwork_uri = (
170        'https://www.googleapis.com/compute/v1/projects/'
171        + self.project_id
172        + '/regions/'
173        + self.region
174        + '/subnetworks/default'
175      )
176    return subnetwork_uri
177
178  @property
179  def is_single_node_cluster(self) -> bool:
180    workers = self._resource_data.get('config', {}).get('workerConfig', {}).get('numInstances', 0)
181    return workers == 0
182
183  @property
184  def is_ha_cluster(self) -> bool:
185    masters = self._resource_data.get('config', {}).get('masterConfig', {}).get('numInstances', 1)
186    return masters != 1
187
188  @property
189  def is_internal_ip_only(self) -> bool:
190    # internalIpOnly is set to true by default when creating a
191    # Dataproc 2.2 image version cluster.
192    # The default should be false in older versions instead.
193    internal_ip_only = self._resource_data['config']['gceClusterConfig']['internalIpOnly']
194    return internal_ip_only
195
196  @property
197  def has_autoscaling_policy(self) -> bool:
198    """Checks if an autoscaling policy is configured for the cluster."""
199    return bool(self._resource_data['config'].get('autoscalingConfig', {}))
200
201  @property
202  def autoscaling_policy_id(self) -> str:
203    """Returns the autoscaling policy ID for the cluster."""
204    if self.has_autoscaling_policy:
205      return (
206        self._resource_data['config']
207        .get('autoscalingConfig', {})
208        .get('policyUri', '')
209        .split('/')[-1]
210      )
211    else:
212      return ''
213
214  @property
215  def number_of_primary_workers(self) -> float:
216    """Gets the number of primary worker nodes in the cluster."""
217    return self._resource_data['config'].get('workerConfig', {}).get('numInstances', 0)
218
219  @property
220  def number_of_secondary_workers(self) -> float:
221    """Gets the number of secondary worker nodes in the cluster."""
222    return self._resource_data['config'].get('secondaryWorkerConfig', {}).get('numInstances', 0)
223
224  @property
225  def is_preemptible_primary_workers(self) -> bool:
226    """Checks if the primary worker nodes in the cluster are preemptible."""
227    return self._resource_data['config'].get('workerConfig', {}).get('isPreemptible', False)
228
229  @property
230  def is_preemptible_secondary_workers(self) -> bool:
231    """Checks if the secondary worker nodes in the cluster are preemptible."""
232    return (
233      self._resource_data['config'].get('secondaryWorkerConfig', {}).get('isPreemptible', False)
234    )
235
236  @property
237  def initialization_actions(self) -> List[str]:
238    return self._resource_data['config'].get('initializationActions', [])

241class Region:
242  """Represents Dataproc region"""
243
244  project_id: str
245  region: str
246
247  def __init__(self, project_id: str, region: str):
248    self.project_id = project_id
249    self.region = region
250
251  def get_clusters(self, context: models.Context) -> Iterable[Cluster]:
252    clusters = []
253    for cluster in self.query_api():
254      if not context.match_project_resource(
255        resource=cluster.get('clusterName'), labels=cluster.get('labels', {})
256      ):
257        continue
258      c = Cluster(
259        name=cluster['clusterName'],
260        project_id=self.project_id,
261        resource_data=cluster,
262      )
263      clusters.append(c)
264    return clusters
265
266  def query_api(self) -> Iterable[dict]:
267    try:
268      api = apis.get_api('dataproc', 'v1', self.project_id)
269      query = (
270        api.projects().regions().clusters().list(projectId=self.project_id, region=self.region)
271      )
272      # be careful not to retry too many times because querying all regions
273      # sometimes causes requests to fail permanently
274      resp = query.execute(num_retries=1)
275      return resp.get('clusters', [])
276    except googleapiclient.errors.HttpError as err:
277      # b/371526148 investigate permission denied error
278      logging.error(err)
279      return []
280      # raise utils.GcpApiError(err) from err

283class Dataproc:
284  """Represents Dataproc product"""
285
286  project_id: str
287
288  def __init__(self, project_id: str):
289    self.project_id = project_id
290
291  def get_regions(self) -> Iterable[Region]:
292    return [Region(self.project_id, r.name) for r in gce.get_all_regions(self.project_id)]
293
294  def is_api_enabled(self) -> bool:
295    return apis.is_enabled(self.project_id, 'dataproc')

298@caching.cached_api_call
299def get_clusters(context: models.Context) -> Iterable[Cluster]:
300  r: List[Cluster] = []
301  dataproc = Dataproc(context.project_id)
302  if not dataproc.is_api_enabled():
303    return r
304  executor = get_executor(context)
305  for clusters in executor.map(lambda r: r.get_clusters(context), dataproc.get_regions()):
306    r += clusters
307  return r

310@caching.cached_api_call
311def get_cluster(cluster_name, region, project) -> Optional[Cluster]:
312  api = apis.get_api('dataproc', 'v1', project)
313  request = (
314    api.projects()
315    .regions()
316    .clusters()
317    .get(projectId=project, clusterName=cluster_name, region=region)
318  )
319  try:
320    r = request.execute(num_retries=config.API_RETRIES)
321  except (googleapiclient.errors.HttpError, requests.exceptions.RequestException):
322    # logging.error(err)
323    return None
324  return Cluster(r['clusterName'], project_id=r['projectId'], resource_data=r)

327class AutoScalingPolicy(models.Resource):
328  """AutoScalingPolicy."""
329
330  _resource_data: dict
331
332  def __init__(self, project_id, resource_data, region):
333    super().__init__(project_id=project_id)
334    self._resource_data = resource_data
335    self.region = region
336
337  @property
338  def policy_id(self) -> str:
339    return self._resource_data['id']
340
341  @property
342  def full_path(self) -> str:
343    return self._resource_data['name']
344
345  @property
346  def short_path(self) -> str:
347    return f'{self.project_id}/{self.region}/{self.policy_id}'
348
349  @property
350  def name(self) -> str:
351    return self._resource_data['name']
352
353  @property
354  def scale_down_factor(self) -> float:
355    return self._resource_data['basicAlgorithm']['yarnConfig'].get('scaleDownFactor', 0.0)
356
357  @property
358  def has_graceful_decommission_timeout(self) -> bool:
359    """Checks if a graceful decommission timeout is configured in the autoscaling policy."""
360    return bool(
361      self._resource_data.get('basicAlgorithm', {})
362      .get('yarnConfig', {})
363      .get('gracefulDecommissionTimeout', {})
364    )
365
366  @property
367  def graceful_decommission_timeout(self) -> float:
368    """Gets the configured graceful decommission timeout in the autoscaling policy."""
369    return (
370      self._resource_data.get('basicAlgorithm', {})
371      .get('yarnConfig', {})
372      .get('gracefulDecommissionTimeout', -1)
373    )

376@caching.cached_api_call
377def get_auto_scaling_policy(project_id: str, region: str, policy_id: str) -> AutoScalingPolicy:
378  logging.debug('fetching autoscalingpolicy: %s', project_id)
379  dataproc = apis.get_api('dataproc', 'v1', project_id)
380  name = f'projects/{project_id}/regions/{region}/autoscalingPolicies/{policy_id}'
381  try:
382    request = dataproc.projects().regions().autoscalingPolicies().get(name=name)
383    response = request.execute(num_retries=config.API_RETRIES)
384    return AutoScalingPolicy(project_id, response, region)
385  except googleapiclient.errors.HttpError as err:
386    raise utils.GcpApiError(err) from err

389@caching.cached_api_call
390def list_auto_scaling_policies(project_id: str, region: str) -> List[AutoScalingPolicy]:
391  """Lists all autoscaling policies in the given project and region."""
392  dataproc = apis.get_api('dataproc', 'v1', project_id)
393  parent = f'projects/{project_id}/regions/{region}'
394  try:
395    request = dataproc.projects().regions().autoscalingPolicies().list(parent=parent)
396    response = request.execute(num_retries=config.API_RETRIES)
397    return [
398      AutoScalingPolicy(project_id, policy_data, region)
399      for policy_data in response.get('policies', [])
400    ]
401  except googleapiclient.errors.HttpError as err:
402    raise utils.GcpApiError(err) from err

405class Job(models.Resource):
406  """Job."""
407
408  _resource_data: dict
409
410  def __init__(self, project_id, job_id, region, resource_data):
411    super().__init__(project_id=project_id)
412    self._resource_data = resource_data
413    self.region = region
414    self.job_id = job_id
415
416  @property
417  def full_path(self) -> str:
418    return f'projects/{self.project_id}/regions/{self.region}/jobs/{self.job_id}'
419
420  @property
421  def short_path(self) -> str:
422    return f'{self.project_id}/{self.region}/{self.job_id}'
423
424  @property
425  def cluster_name(self) -> str:
426    return self._resource_data['placement']['clusterName']
427
428  @property
429  def cluster_uuid(self) -> str:
430    return self._resource_data['placement']['clusterUuid']
431
432  @property
433  def state(self):
434    return self._resource_data['status']['state']
435
436  @property
437  def details(self):
438    if self._resource_data['status']['state'] == 'ERROR':
439      return self._resource_data['status']['details']
440    return None
441
442  @property
443  def status_history(self):
444    status_history_dict = {}
445    for previous_status in self._resource_data['statusHistory']:
446      if previous_status['state'] not in status_history_dict:
447        status_history_dict[previous_status['state']] = previous_status['stateStartTime']
448
449    return status_history_dict
450
451  @property
452  def yarn_applications(self):
453    return self._resource_data['yarnApplications']
454
455  @property
456  def driver_output_resource_uri(self):
457    return self._resource_data.get('driverOutputResourceUri')
458
459  @property
460  def job_uuid(self):
461    return self._resource_data.get('jobUuid')
462
463  @property
464  def job_provided_bq_connector(self):
465    """Check user-supplied BigQuery connector on the job level"""
466    jar_file_uris = self._resource_data.get('sparkJob', {}).get('jarFileUris')
467    if jar_file_uris is not None:
468      for file in jar_file_uris:
469        if 'spark-bigquery-latest.jar' in file:
470          return 'spark-bigquery-latest'
471        else:
472          match = re.search(
473            r'spark-bigquery(?:-with-dependencies_\d+\.\d+)?-(\d+\.\d+\.\d+)\.jar', file
474          )
475          if match:
476            return match.group(1)
477    return None

480@caching.cached_api_call
481def get_job_by_jobid(project_id: str, region: str, job_id: str):
482  dataproc = apis.get_api('dataproc', 'v1', project_id)
483  try:
484    request = (
485      dataproc.projects().regions().jobs().get(projectId=project_id, region=region, jobId=job_id)
486    )
487    response = request.execute(num_retries=config.API_RETRIES)
488    return Job(project_id, region, job_id, response)
489  except googleapiclient.errors.HttpError as err:
490    raise utils.GcpApiError(err) from err

493@caching.cached_api_call
494def extract_dataproc_supported_version() -> list[str]:
495  """Extract the supported Dataproc versions(use Debian as representative)."""
496
497  page_url = 'https://cloud.google.com/dataproc/docs/concepts/versioning/dataproc-version-clusters'
498
499  try:
500    table = web.fetch_and_extract_table(page_url, tag='h3', tag_id='debian_images')
501    if table:
502      rows = table.find_all('tr')[1:]  # Skip the header row
503      version_list = []
504
505      for row in rows:
506        dp_version = row.find_all('td')[0].get_text().strip().split('-')[0]
507        version_list.append(dp_version)
508      return version_list
509
510    else:
511      return []
512  except (
513    requests.exceptions.RequestException,
514    AttributeError,
515    TypeError,
516    ValueError,
517    IndexError,
518  ) as e:
519    logging.error(
520      'Error in extracting dataproc versions: %s',
521      e,
522    )
523    return []

526@caching.cached_api_call
527def extract_dataproc_bigquery_version(image_version) -> list[str]:
528  """Extract Dataproc BigQuery connector versions based on image version GCP documentation."""
529
530  page_url = (
531    'https://cloud.google.com/dataproc/docs/concepts/versioning/dataproc-release-' + image_version
532  )
533
534  try:
535    table = web.fetch_and_extract_table(page_url, tag='div')
536    bq_version = []
537    if table:
538      rows = table.find_all('tr')[1:]
539      for row in rows:
540        cells = row.find_all('td')
541        if 'BigQuery Connector' in cells[0].get_text(strip=True):
542          bq_version = cells[1].get_text(strip=True)
543    return bq_version
544  except (
545    requests.exceptions.RequestException,
546    AttributeError,
547    TypeError,
548    ValueError,
549    IndexError,
550  ) as e:
551    logging.error(
552      '%s Error in extracting BigQuery connector versions.'
553      '  Please check BigQuery Connector version on %s',
554      e,
555      page_url,
556    )
557    return []