Ignore timeouts that occur while sampling nodes, replicas.

This could result in gaps in the data, but that's almost always ok.

PiperOrigin-RevId: 721487361

Author: p3rf Team

perfkitbenchmarker/container_service.py

@@ -130,6 +130,7 @@
     ),
     'read: connection reset by peer',
     'Unable to connect to the server: dial tcp',
+    'Unable to connect to the server: net/http: TLS handshake timeout',
 ]
 
 
@@ -161,6 +162,36 @@ def RunKubectlCommand(command: list[str], **kwargs) -> tuple[str, str, int]:
     # IssueCommand defaults stack_level to 1, so 2 skips this function.
     kwargs['stack_level'] = 2
   cmd = [FLAGS.kubectl, '--kubeconfig', FLAGS.kubeconfig] + command
+
+  orig_suppress_failure = None
+  if 'suppress_failure' in kwargs:
+    orig_suppress_failure = kwargs['suppress_failure']
+
+  def _DetectTimeoutViaSuppressFailure(stdout, stderr, retcode):
+    # Check for kubectl timeout. If found, treat it the same as a regular
+    # timeout.
+    if retcode != 0:
+      for error_substring in _RETRYABLE_KUBECTL_ERRORS:
+        if error_substring in stderr:
+          # Raise timeout error regardless of raise_on_failure - as the intended
+          # semantics is to ignore expected errors caused by invoking the
+          # command not errors from PKB infrastructure.
+          raise_on_timeout = (
+              kwargs['raise_on_timeout']
+              if 'raise_on_timeout' in kwargs
+              else True
+          )
+          if raise_on_timeout:
+            raise errors.VmUtil.IssueCommandTimeoutError(stderr)
+    # Else, if the user supplied a suppress_failure function, try that.
+    if orig_suppress_failure is not None:
+      return orig_suppress_failure(stdout, stderr, retcode)
+
+    # Else, no suppression.
+    return False
+
+  kwargs['suppress_failure'] = _DetectTimeoutViaSuppressFailure
+
   return vm_util.IssueCommand(cmd, **kwargs)
 
 
@@ -172,6 +203,12 @@ def RunRetryableKubectlCommand(
     run_cmd: list[str], timeout: int | None = None, **kwargs
 ) -> tuple[str, str, int]:
   """Runs a kubectl command, retrying somewhat exepected errors."""
+  if 'raise_on_timeout' in kwargs and kwargs['raise_on_timeout']:
+    raise ValueError(
+        'RunRetryableKubectlCommand does not allow `raise_on_timeout=True`'
+        ' (since timeouts are retryable).'
+    )
+
   if 'stack_level' in kwargs:
     kwargs['stack_level'] += 1
   else:
@@ -180,24 +217,12 @@ def RunRetryableKubectlCommand(
 
   @vm_util.Retry(
       timeout=timeout,
-      retryable_exceptions=(RetryableKubectlError,),
+      retryable_exceptions=(errors.VmUtil.IssueCommandTimeoutError,),
   )
   def _RunRetryablePart(run_cmd: list[str], **kwargs):
     """Inner function retries command so timeout can be passed to decorator."""
     kwargs['stack_level'] += 1
-    out, err, code = RunKubectlCommand(
-        run_cmd, raise_on_failure=False, **kwargs
-    )
-    if err:
-      logging.warning('Got error %s when running %s', err, run_cmd)
-      for error_substring in _RETRYABLE_KUBECTL_ERRORS:
-        if error_substring in err:
-          raise RetryableKubectlError(
-              f'Tried running {run_cmd} but it failed with the substring'
-              f' {error_substring}. Retrying. Full error is: {err}'
-          )
-      raise errors.VmUtil.IssueCommandError(err)
-    return out, err, code
+    return RunKubectlCommand(run_cmd, raise_on_timeout=True, **kwargs)
 
   return _RunRetryablePart(run_cmd, timeout=timeout, **kwargs)
 

perfkitbenchmarker/linux_benchmarks/kubernetes_hpa_benchmark.py

@@ -24,6 +24,7 @@
 from perfkitbenchmarker import benchmark_spec as bm_spec
 from perfkitbenchmarker import configs
 from perfkitbenchmarker import container_service
+from perfkitbenchmarker import errors
 from perfkitbenchmarker.linux_packages import locust
 from perfkitbenchmarker.sample import Sample
 
@@ -226,8 +227,20 @@ def _Observe(
       self,
       observe_fn: Callable[[], List[Sample]],
   ) -> None:
+    """Call the specified function until self._stop is signalled.
+
+    Results are appended to self._samples. Timeouts are ignored.
+
+    Args:
+      observe_fn: The function to call.
+    """
     while True:
-      self._samples.extend(observe_fn())
+      try:
+        self._samples.extend(observe_fn())
+      except errors.VmUtil.IssueCommandTimeoutError:
+        # Ignore timeouts - there'll be a gap in the data, but that's ok.
+        pass
+
       if self._stop.wait(timeout=1.0):
         return
 

tests/container_service_test.py

@@ -1,10 +1,12 @@
-from typing import Iterable
+import time
+from typing import Callable, Iterable, Protocol, Tuple
 import unittest
 from unittest import mock
 from absl.testing import flagsaver
 from perfkitbenchmarker import container_service
 from perfkitbenchmarker import errors
 from perfkitbenchmarker import provider_info
+from perfkitbenchmarker import vm_util
 from perfkitbenchmarker.configs import container_spec
 from perfkitbenchmarker.sample import Sample
 from tests import pkb_common_test_case
@@ -25,6 +27,60 @@ def _Delete(self):
     pass
 
 
+kubectl_timeout_tuple = (
+    '',
+    (
+        'Unable to connect to the server: dial tcp 10.42.42.42:443:'
+        'connect: connection timed out'
+    ),
+    1,
+)
+
+
+class _IssueCommandCallable(Protocol):
+
+  def __call__(
+      self,
+      cmd: Iterable[str],
+      suppress_failure: Callable[[str, str, int], bool] | None = None,
+      **kwargs,
+  ) -> Tuple[str, str, int]:
+    ...
+
+
+def _MockedIssueCommandSuppressing(
+    stderr: str,
+) -> _IssueCommandCallable:
+  def _MockedCommand(
+      cmd: Iterable[str],
+      suppress_failure: Callable[[str, str, int], bool] | None = None,
+      **kwargs,
+  ):
+    _ = cmd
+    _ = kwargs
+    stdout = ''
+    status = 1
+    if suppress_failure and suppress_failure(stdout, stderr, status):
+      return stdout, '', 0
+    return stdout, stderr, status
+
+  return _MockedCommand
+
+
+def _MockedIssueCommandFailure(
+    cmd: Iterable[str],
+    suppress_failure: Callable[[str, str, int], bool] | None = None,
+    **kwargs,
+) -> Tuple[str, str, int]:
+  return _MockedIssueCommandSuppressing(
+      stderr='A failure occurred',
+  )(
+      cmd,
+      suppress_failure=suppress_failure,
+      **kwargs,
+  )
+
+
 class ContainerServiceTest(pkb_common_test_case.PkbCommonTestCase):
 
   def setUp(self):
@@ -62,27 +118,60 @@ def test_apply_manifest_gets_deployment_name(self):
     )
     self.assertEqual(next(deploy_ids), 'deployment.apps/test-deployment')
 
-  def test_retriable_kubectl_command_fails_on_random_error(self):
-    self.MockIssueCommand(
-        {'get podpatchwork': [('', 'error: invalid syntax', 1)]}
-    )
+  @mock.patch.object(
+      vm_util,
+      'IssueCommand',
+      side_effect=[errors.VmUtil.IssueCommandError()],
+      autospec=True,
+  )
+  def test_retriable_kubectl_command_fails_on_random_error(self, _):
     with self.assertRaises(errors.VmUtil.IssueCommandError):
       container_service.RunRetryableKubectlCommand(['get', 'podpatchwork'])
 
-  def test_retriable_kubectl_command_retries_on_connection_reset(self):
-    self.MockIssueCommand({
-        'get pods': [
-            ('', 'error: read: connection reset by peer', 1),
-            ('pod1, pod2', '', 0),
-        ]
-    })
+  @mock.patch.object(
+      vm_util,
+      'IssueCommand',
+      side_effect=[
+          errors.VmUtil.IssueCommandTimeoutError(),
+          ('pod1, pod2', '', 0),
+      ],
+      autospec=True,
+  )
+  @mock.patch.object(time, 'sleep', autospec=True)
+  def test_retriable_kubectl_command_retries_on_retriable_error(
+      self, sleep_mock, issue_command_mock
+  ):
     out, err, ret = container_service.RunRetryableKubectlCommand(
         ['get', 'pods']
     )
     self.assertEqual(out, 'pod1, pod2')
     self.assertEqual(err, '')
     self.assertEqual(ret, 0)
 
+  def test_retriable_kubectl_command_passes_timeout_through(self):
+    def _VerifyTimeout(
+        cmd: Iterable[str],
+        timeout: int | None = vm_util.DEFAULT_TIMEOUT,
+        **kwargs,
+    ) -> Tuple[str, str, int]:
+      _ = cmd
+      _ = kwargs
+      self.assertEqual(
+          timeout,
+          1,
+          'timeout not correctly passed to underlying vm_util.IssueCommand()',
+      )
+      return 'ok', '', 0
+
+    with mock.patch.object(vm_util, 'IssueCommand', _VerifyTimeout):
+      container_service.RunRetryableKubectlCommand(['get', 'pods'], timeout=1)
+
+  def test_retriable_kubectl_command_fails_with_raise_on_timeout(self):
+    with self.assertRaises(ValueError):
+      container_service.RunRetryableKubectlCommand(
+          ['get', 'pods'], raise_on_timeout=True
+      )
+
   def test_GetNumReplicasSamples_found(self):
     resource_name = 'deployment/my_deployment'
     namespace = 'my_namespace'
@@ -161,6 +250,63 @@ def _Sample(count: int, state: str) -> Sample:
         ],
     )
 
+  @mock.patch.object(
+      vm_util,
+      'IssueCommand',
+      return_value=['stdout', 'stderr', 0],
+      autospec=True,
+  )
+  def test_RunKubectlCommand(self, issue_command_mock):
+    stdout, stderr, status = container_service.RunKubectlCommand(
+        ['get', 'pods']
+    )
+    self.assertEqual(stdout, 'stdout')
+    self.assertEqual(stderr, 'stderr')
+    self.assertEqual(status, 0)
+
+  @mock.patch.object(
+      vm_util,
+      'IssueCommand',
+      side_effect=errors.VmUtil.IssueCommandTimeoutError(),
+      autospec=True,
+  )
+  def test_RunKubectlCommand_CommandTimeoutPropagated(self, issue_command_mock):
+    with self.assertRaises(errors.VmUtil.IssueCommandTimeoutError):
+      container_service.RunKubectlCommand(['get', 'pods'])
+
+  def test_RunKubectlCommand_KubectlTimeoutRaisesCommandTimeout(self):
+    for err in container_service._RETRYABLE_KUBECTL_ERRORS:
+      with mock.patch.object(
+          vm_util, 'IssueCommand', _MockedIssueCommandSuppressing(stderr=err)
+      ):
+        with self.assertRaises(
+            errors.VmUtil.IssueCommandTimeoutError,
+            msg=f'Failed to raise timeout for error: {err}',
+        ):
+          container_service.RunKubectlCommand(['get', 'pods'])
+
+  def test_RunKubectlCommand_KubectlTimeoutWithSuppressFailureRaisesCommandTimeout(
+      self,
+  ):
+    for err in container_service._RETRYABLE_KUBECTL_ERRORS:
+      with mock.patch.object(
+          vm_util, 'IssueCommand', _MockedIssueCommandSuppressing(stderr=err)
+      ):
+        with self.assertRaises(
+            errors.VmUtil.IssueCommandTimeoutError,
+            msg=f'Failed to raise timeout for error: {err}',
+        ):
+          container_service.RunKubectlCommand(
+              ['get', 'pods'], suppress_failure=lambda x, y, z: True
+          )
+
+  @mock.patch.object(vm_util, 'IssueCommand', _MockedIssueCommandFailure)
+  def test_RunKubectlCommand_KubectlFailWithSuppressFailure(self):
+    _, _, status = container_service.RunKubectlCommand(
+        ['get', 'pods'], suppress_failure=lambda x, y, z: True
+    )
+    self.assertEqual(status, 0)
+
 
 def _ClearTimestamps(samples: Iterable[Sample]) -> Iterable[Sample]:
   for s in samples: