Merge branch 'GoogleCloudPlatform:master' into swap-encryption-pr1-shared-infra_new

Author: DevVegeta

CHANGES.next.md

@@ -127,6 +127,7 @@
     bringing this variant more in-line with others.
 -   Rename GKE Autopilot & EKS Auto mode's cluster_type within PKB from
     'Autopilot' to 'Auto'.
+-   Duplicate metadata for all samples.
 
 ### New features:
 
@@ -469,6 +470,7 @@
     `--use_managed_vm_groups`.
 -   Add support for configuring IMDSv2 Http Tokens on AWS VMs via
     `aws_metadata_http_tokens`.
+-   Add aggregate memtier results to `kubernetes_redis_memtier` benchmark.
 
 ### Bug fixes and maintenance updates:
 

perfkitbenchmarker/linux_benchmarks/cluster_boot_benchmark.py

@@ -168,7 +168,7 @@
 )
 _COLLECT_BTIME = flags.DEFINE_boolean(
     'cluster_boot_collect_btime',
-    True,
+    False,
     'Collect btime from /proc/stat. This should be the same as kernel_start '
     'from linux_boot, but does not require a startup script.',
 )

perfkitbenchmarker/linux_benchmarks/kubernetes_redis_memtier_benchmark.py

@@ -213,7 +213,9 @@ def _RunMemtier(
       pipeline=pipeline,
   )
 
-  return result.GetSamples(metadata)
+  samples = result.GetSamples(metadata)
+  samples.extend(memtier.AggregateMemtierResults([result], metadata))
+  return samples
 
 
 class MemtierRunConfig(NamedTuple):

perfkitbenchmarker/linux_packages/fortran.py

@@ -15,6 +15,7 @@
 
 import logging
 from absl import flags
+from perfkitbenchmarker import errors
 
 VERSION = flags.DEFINE_integer(
     'fortran_version', None, 'Version of gfortran to install'
@@ -33,7 +34,7 @@ def YumInstall(vm):
     _YumInstallVersion(vm, VERSION.value)
   else:
     vm.InstallPackages('gcc-gfortran libgfortran')
-  _LogFortranVersion(vm)
+  logging.info('Version of fortran: %s', GetFortranVersion(vm))
 
 
 def AptInstall(vm):
@@ -42,7 +43,7 @@ def AptInstall(vm):
     _AptInstallVersion(vm, VERSION.value)
   else:
     vm.InstallPackages('gfortran')
-  _LogFortranVersion(vm)
+  logging.info('Version of fortran: %s', GetFortranVersion(vm))
 
 
 def _YumInstallVersion(vm, version):
@@ -73,7 +74,18 @@ def _AptInstallVersion(vm, version):
   )
 
 
-def _LogFortranVersion(vm):
-  """Logs the version of gfortran."""
-  txt, _ = vm.RemoteCommand('gfortran -dumpversion')
-  logging.info('Version of fortran: %s', txt.strip())
+def GetFortranVersion(vm, required_version_prefix: str | None = None) -> str:
+  """Returns the full version of gfortran e.g. 11.5.0."""
+  txt, _ = vm.RemoteCommand('gfortran -dumpfullversion')
+  version = txt.strip()
+  if not required_version_prefix and VERSION.value:
+    required_version_prefix = str(VERSION.value)
+
+  if required_version_prefix and not version.startswith(
+      required_version_prefix
+  ):
+    raise errors.Benchmarks.PrepareException(
+        f'gfortran version {version} does not match expected version'
+        f' {required_version_prefix}'
+    )
+  return version

perfkitbenchmarker/linux_packages/speccpu.py

@@ -26,6 +26,7 @@
 from perfkitbenchmarker import sample
 from perfkitbenchmarker import stages
 from perfkitbenchmarker.linux_packages import build_tools
+from perfkitbenchmarker.linux_packages import fortran
 
 BASE_MODE = 'base'
 PEAK_MODE = 'peak'
@@ -290,9 +291,10 @@ def InstallSPECCPU(vm, speccpu_vm_state):
 def Install(vm):
   """Installs SPECCPU dependencies."""
   vm.Install('wget')
-  vm.Install('fortran')
   # Install gcc, g++, fortran according to --gcc_version
   vm.Install('build_tools')
+  # Verify fortran is installed correctly
+  fortran.GetFortranVersion(vm, required_version_prefix=FLAGS.gcc_version)
 
   # Install libxcrypt-compat is needed on RHEL 9 based OSes, but not available
   # on RHEL 8 based OSes.
@@ -505,6 +507,7 @@ def _ExtractScore(stdout, vm, keep_partial_results, runspec_metric):
       'spec17_fdo': FLAGS.spec17_fdo,
       'spec17_subset': FLAGS.spec17_subset,
       'gcc_version': build_tools.GetVersion(vm, 'gcc'),
+      'gfortran_version': fortran.GetFortranVersion(vm),
   }
   if cpu_version:
     major, minor = cpu_version
@@ -551,12 +554,15 @@ def _ExtractScore(stdout, vm, keep_partial_results, runspec_metric):
     missing_results.append(spec_name)
 
   if missing_results:
+    missing_results_str = ','.join(missing_results)
     if keep_partial_results:
       metadata['partial'] = 'true'
-      metadata['missing_results'] = ','.join(missing_results)
+      metadata['missing_results'] = missing_results_str
+      for result in results:
+        result.metadata.update(metadata)
     else:
       raise errors.Benchmarks.RunError(
-          'speccpu: results missing, see log: ' + ','.join(missing_results)
+          'speccpu: results missing, see log: ' + missing_results_str
       )
 
   if spec_score:

perfkitbenchmarker/providers/gcp/gce_virtual_machine.py

@@ -1077,6 +1077,11 @@ def _ParseCreateErrors(
       # Occurs when creating a large number >30 of VMs in parallel.
       if 'gcloud crashed (SSLError)' in stderr:
         raise errors.Resource.RetryableCreationError(stderr)
+      if (
+          'The global or regional External IP resource pool is exhausted'
+          in stderr
+      ):
+        raise errors.Benchmarks.InsufficientCapacityCloudFailure(stderr)
       raise errors.Resource.CreationError(
           f'Failed to create VM {self.name}:\n{stderr}\nreturn code: {retcode}'
       )

perfkitbenchmarker/sample.py

@@ -136,12 +136,14 @@ def __new__(
     if timestamp is None:
       timestamp = time.time()
 
+    metadata = metadata or {}
+    metadata = metadata.copy()
     return super().__new__(
         cls,
         metric,
         float(value or 0.0),
         unit,
-        metadata=metadata or {},
+        metadata=metadata,
         timestamp=timestamp,
         **kwargs,
     )

tests/gce_virtual_machine_test.py

@@ -1104,6 +1104,14 @@ def testCreateVMSubnetworkNotReady(self):
           ),
           'expected_error': errors.Resource.RetryableCreationError,
       },
+      {
+          'testcase_name': 'IP stockout',
+          'fake_stderr': (
+              '{"errors": [{"code": "RESOURCE_POOL_EXHAUSTED", "message": "The'
+              ' global or regional External IP resource pool is exhausted"}]}'
+          ),
+          'expected_error': errors.Benchmarks.InsufficientCapacityCloudFailure,
+      }
   )
   def testCreateVMErrorCases(self, fake_stderr, expected_error):
     fake_rets = [('stdout', fake_stderr, 1)]

tests/linux_benchmarks/cluster_boot_benchmark_test.py

@@ -216,13 +216,6 @@ def testGetTimeToBoot(
             vm_util, 'IssueCommand', return_value=('{}', '', 0), autospec=True
         )
     )
-    self.enter_context(
-        mock.patch(
-            'perfkitbenchmarker.linux_packages.linux_boot.CollectBtimeSample',
-            return_value=sample.Sample('btime', 5.0, 'second', {}),
-            autospec=True,
-        )
-    )
     vm_spec = vm_spec_class(
         'cluster_boot_benchmark_test', zone=zone, machine_type=machine_type
     )
@@ -272,16 +265,6 @@ def testGetTimeToBoot(
             timestamp=1678147200.0,
         )
     )
-    expecteds.append(
-        sample.Sample(
-            metric='btime',
-            value=5.0,
-            unit='second',
-            metadata={},
-            timestamp=1678147200.0,
-        )
-    )
-
     self.assertCountEqual(actuals, expecteds)
 
 

tests/linux_benchmarks/dpb_sparksql_benchmark_test.py

@@ -18,7 +18,6 @@
 
 from perfkitbenchmarker.scripts.spark_sql_test_scripts import spark_sql_runner
 
-
 FLAGS = flags.FLAGS
 _TPCH_TABLES = [
     'customer',
@@ -391,6 +390,52 @@ def testRenderRunnerScriptWithQueries(self):
     )
     self.assertIn(expected_substr, rendered_script)
 
+  @flagsaver.flagsaver(
+      dpb_sparksql_order=['1', '2', '3', '4'],
+      dpb_sparksql_fetch_results_from_logs=True,
+  )
+  def testSampleFailingQueries(self):
+    # Arrange
+    self.benchmark_spec_mock.table_subdirs = list(_TPCH_TABLES)
+    self.benchmark_spec_mock.query_streams = [['1', '2', '3']]
+    self.benchmark_spec_mock.queries_and_script_upload_time = None
+    self.benchmark_spec_mock.hive_tables_creation_time = None
+
+    job_result_mock = mock.Mock(
+        wall_time=42.0,
+        pending_time=5.0,
+        stdout=(
+            '----@spark_sql_runner:results_start@----\n'
+            '[{"query_id": "1", "duration": 10.0},\n'
+            ' {"query_id": "2", "duration": 10.0},\n'
+            ' {"query_id": "3", "duration": 10.0}]\n'
+            '----@spark_sql_runner:results_end@----\n'
+        ),
+        stderr='',
+    )
+    self.benchmark_spec_mock.dpb_service.SubmitJob.return_value = (
+        job_result_mock
+    )
+    self.benchmark_spec_mock.dpb_service.GetResourceMetadata.return_value = {
+        'cluster_name': 'test_cluster'
+    }
+    self.benchmark_spec_mock.dpb_service.GetClusterCreateTime.return_value = (
+        100.0
+    )
+    self.benchmark_spec_mock.dpb_service.metadata = {}
+
+    # Act
+    samples = dpb_sparksql_benchmark.Run(self.benchmark_spec_mock)
+
+    # Assert
+    query_sample = next(s for s in samples if s.metric == 'sparksql_run_time')
+    global_sample = next(
+        s for s in samples if s.metric == 'sparksql_total_wall_time'
+    )
+    self.assertIn('failing_queries', global_sample.metadata)
+    self.assertEqual(global_sample.metadata['failing_queries'], '4')
+    self.assertNotIn('failing_queries', query_sample.metadata)
+
 
 if __name__ == '__main__':
   unittest.main()