Separate service-reported and end-to-end cluster creation times.

PiperOrigin-RevId: 930777510

Author: dorellang

perfkitbenchmarker/dpb_service.py

@@ -681,6 +681,15 @@ def GetClusterCreateTime(self) -> float | None:
       return None
     return self.resource_ready_time - self.create_start_time
 
+  def GetServiceReportedClusterCreateTime(self) -> float | None:
+    """Returns the cluster creation time as reported by the service.
+
+    Returns:
+      A float representing the creation time in seconds or None if not
+      implemented.
+    """
+    return None
+
   def GetServiceWrapperScriptsToUpload(self) -> List[str]:
     """Gets service wrapper scripts to upload alongside benchmark scripts."""
     return []
@@ -756,9 +765,13 @@ def GetSamples(self) -> list[sample.Sample]:
     """Gets samples with service statistics."""
     samples = []
     metrics: dict[str, tuple[float | None, str]] = {
-        # Cluster creation time as reported by the DPB service
-        # (non-Serverless DPB services only).
+        # Cluster creation time as measured on PKB's end (E2E).
         'dpb_cluster_create_time': (self.GetClusterCreateTime(), 'seconds'),
+        # Cluster creation time as reported by the DPB service backend.
+        'dpb_service_reported_cluster_create_time': (
+            self.GetServiceReportedClusterCreateTime(),
+            'seconds',
+        ),
         # Cluster duration as computed by the underlying benchmark.
         # (non-Serverless DPB services only).
         'dpb_cluster_duration': (self.GetClusterDuration(), 'seconds'),
@@ -805,6 +818,9 @@ def _Delete(self) -> None:
   def GetClusterCreateTime(self) -> float | None:
     return None
 
+  def GetServiceReportedClusterCreateTime(self) -> float | None:
+    return None
+
   def GetClusterDuration(self) -> float | None:
     return None
 

perfkitbenchmarker/providers/aws/aws_dpb_emr.py

@@ -22,6 +22,7 @@
 import json
 import logging
 import os
+import time
 from typing import Any, Dict
 
 from absl import flags
@@ -39,7 +40,6 @@
 from perfkitbenchmarker.providers.aws import s3
 from perfkitbenchmarker.providers.aws import util
 
-
 FLAGS = flags.FLAGS
 flags.DEFINE_string(
     'dpb_emr_release_label', None, 'DEPRECATED use dpb_service.version.'
@@ -197,8 +197,8 @@ def __init__(self, dpb_service_spec):
     self.storage_service = s3.S3Service()
     self.storage_service.PrepareService(self.region)
     self.bucket_to_delete = None
-    self._cluster_create_time: float | None = None
-    self._cluster_ready_time: float | None = None
+    self._service_reported_cluster_create_time: float | None = None
+    self._service_reported_cluster_ready_time: float | None = None
     self._cluster_delete_time: float | None = None
     if not self.GetDpbVersion():
       raise errors.Setup.InvalidSetupError(
@@ -212,17 +212,26 @@ def persistent_fs_prefix(self) -> str | None:
   def GetDpbVersion(self) -> str | None:
     return FLAGS.dpb_emr_release_label or super().GetDpbVersion()
 
-  def GetClusterCreateTime(self) -> float | None:
-    """Returns the cluster creation time.
+  def GetServiceReportedClusterCreateTime(self) -> float | None:
+    """Returns the cluster creation time as reported by the service.
 
     On this implementation, the time returned is based on the timestamps
-    reported by the EMR API (which is stored in the _cluster_create_time
+    reported by the EMR API (which is stored in the
+    _service_reported_cluster_create_time
     attribute).
 
     Returns:
       A float representing the creation time in seconds or None.
     """
-    return self._cluster_ready_time - self._cluster_create_time
+    if (
+        self._service_reported_cluster_create_time is None
+        or self._service_reported_cluster_ready_time is None
+    ):
+      return None
+    return (
+        self._service_reported_cluster_ready_time
+        - self._service_reported_cluster_create_time
+    )
 
   @property
   def security_group_id(self):
@@ -397,9 +406,11 @@ def _IsReady(self):
     # TODO(saksena): Handle error outcomees when spinning up emr clusters
     is_ready = result['Cluster']['Status']['State'] == READY_STATE
     if is_ready:
-      self._cluster_create_time, self._cluster_ready_time = (
-          self._ParseClusterCreateTime(result)
-      )
+      (
+          self._service_reported_cluster_create_time,
+          self._service_reported_cluster_ready_time,
+      ) = self._ParseClusterCreateTime(result)
+      self.resource_ready_time = time.time()
     return is_ready
 
   @classmethod
@@ -575,8 +586,7 @@ def _FetchLogs(self, step_id: str) -> str | None:
         's3',
         'cp',
         os.path.join(
-            self.base_dir,
-            f'{self.cluster_id}/steps/{step_id}/stderr.gz'
+            self.base_dir, f'{self.cluster_id}/steps/{step_id}/stderr.gz'
         ),
         local_stdout_path,
     ]
@@ -615,7 +625,7 @@ def __init__(self, dpb_service_spec):
     self.cmd_prefix += ['--region', self.region]
     self.storage_service = s3.S3Service()
     self.storage_service.PrepareService(self.region)
-    self._cluster_create_time = None
+    self._service_reported_cluster_create_time = None
     if not self.GetDpbVersion():
       raise errors.Setup.InvalidSetupError(
           'dpb_service.version must be provided. Versions follow the format: '

perfkitbenchmarker/providers/gcp/gcp_dpb_dataproc.py

@@ -22,6 +22,7 @@
 import logging
 import os
 import re
+import time
 from typing import Any, Dict, Tuple
 
 from absl import flags
@@ -88,8 +89,8 @@ def __init__(self, dpb_service_spec):
     self.region = self.dpb_service_zone.rsplit('-', 1)[0]
     self.storage_service = gcs.GoogleCloudStorageService()
     self.storage_service.PrepareService(location=self.region)
-    self._cluster_create_time: float | None = None
-    self._cluster_ready_time: float | None = None
+    self._service_reported_cluster_create_time: float | None = None
+    self._service_reported_cluster_ready_time: float | None = None
     self._cluster_delete_time: float | None = None
 
   @property
@@ -179,19 +180,26 @@ def _InitializeMetadata(self) -> None:
         gcp_flags.GCP_DATAPROC_LIGHTNING_ENGINE_RUNTIME.value
     )
 
-  def GetClusterCreateTime(self) -> float | None:
-    """Returns the cluster creation time.
+  def GetServiceReportedClusterCreateTime(self) -> float | None:
+    """Returns the cluster creation time as reported by the service.
 
     On this implementation, the time returned is based on the timestamps
-    reported by the Dataproc API (which is stored in the _cluster_create_time
+    reported by the Dataproc API (which is stored in the
+    _service_reported_cluster_create_time
     attribute).
 
     Returns:
       A float representing the creation time in seconds or None.
     """
-    if self._cluster_create_time is None or self._cluster_ready_time is None:
+    if (
+        self._service_reported_cluster_create_time is None
+        or self._service_reported_cluster_ready_time is None
+    ):
       return None
-    return self._cluster_ready_time - self._cluster_create_time
+    return (
+        self._service_reported_cluster_ready_time
+        - self._service_reported_cluster_create_time
+    )
 
   def _Create(self):
     """Creates the cluster."""
@@ -293,13 +301,16 @@ def _Create(self):
     cmd.flags['labels'] = util.MakeFormattedDefaultTags()
     timeout = 900  # 15 min
     stdout, stderr, retcode = cmd.Issue(timeout=timeout, raise_on_failure=False)
-    self._cluster_create_time, self._cluster_ready_time = (
-        self._ParseClusterCreateTime(stdout)
-    )
+    (
+        self._service_reported_cluster_create_time,
+        self._service_reported_cluster_ready_time,
+    ) = self._ParseClusterCreateTime(stdout)
     if retcode:
       util.CheckGcloudResponseKnownFailures(stderr, retcode)
       raise errors.Resource.CreationError(stderr)
 
+    self.resource_ready_time = time.time()
+
   @classmethod
   def _ParseClusterCreateTime(
       cls, stdout: str