diff --git a/perfkitbenchmarker/linux_benchmarks/kubernetes_deployment_startup_benchmark.py b/perfkitbenchmarker/linux_benchmarks/kubernetes_deployment_startup_benchmark.py index 5a444dbe3c..a4cfc1a73b 100644 --- a/perfkitbenchmarker/linux_benchmarks/kubernetes_deployment_startup_benchmark.py +++ b/perfkitbenchmarker/linux_benchmarks/kubernetes_deployment_startup_benchmark.py @@ -11,18 +11,46 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -"""Benchmark for measuring time to start up a deployment on Kubernetes.""" +"""Benchmark for measuring time to start up a deployment on Kubernetes. + +PR 1 — Metrics & Observability +================================ +Extends the existing benchmark with two new metrics and per-sample metadata: + +1. **per_pod_ready_time** — individual pod startup duration (s) from + PodReadyToStartContainers → Ready, using the existing + kubernetes_conditions.GetStatusConditionsForResourceType() call that + already powers max_pod_ready_time. Emits one Sample per pod so + downstream analysis can compute percentiles across replicas. + +2. **cpu_utilization_millicores** (peak / mean / count) — CPU sampled in a + background thread via KubernetesMetricsCollector._Observe(), following + the exact same pattern as kubernetes_hpa_benchmark.py. + +3. **metadata** — scenario / workload / cloud / replicas added to every + sample for cross-config comparison (PR 3 will set scenario=optimized). + +Nothing in Prepare() or Cleanup() changes. PR 2 adds vLLM workload +support; PR 3 adds the VPA CPU Startup Boost scenario flag. +""" import collections +import logging +import threading +from collections.abc import Callable from typing import Any, Dict, List from absl import flags from perfkitbenchmarker import benchmark_spec as bm_spec from perfkitbenchmarker import configs +from perfkitbenchmarker import errors from perfkitbenchmarker import sample from perfkitbenchmarker.resources.container_service import kubernetes_commands +from perfkitbenchmarker.resources.container_service import kubectl from perfkitbenchmarker.resources.container_service import kubernetes_conditions +FLAGS = flags.FLAGS + BENCHMARK_NAME = 'kubernetes_deployment_startup' BENCHMARK_CONFIG = """ kubernetes_deployment_startup: @@ -54,8 +82,28 @@ 'Image name. If omitted, "slowjvmstartup" will be used', ) +# PR 1: new flags — workload and scenario are stubs here; PR 2 and PR 3 add +# the actual logic so that flag names are stable across all three PRs. +WORKLOAD = flags.DEFINE_enum( + 'kubernetes_deployment_startup_workload', + 'jvm', + ['jvm', 'vllm'], + 'Workload type. vLLM deployment support is added in PR 2.', +) +SCENARIO = flags.DEFINE_enum( + 'kubernetes_deployment_startup_scenario', + 'baseline', + ['baseline', 'optimized'], + 'Startup scenario. optimized (GKE CPU Startup Boost) is added in PR 3.', +) + +# Interval between successive CPU polls (seconds). Matches +# kubernetes_hpa_benchmark which polls at ~1 s. +_CPU_POLL_INTERVAL_SECS = 5 + def GetConfig(user_config: Dict[str, Any]) -> Dict[str, Any]: + """Returns merged benchmark config.""" config = configs.LoadConfig(BENCHMARK_CONFIG, user_config, BENCHMARK_NAME) if DEPLOYMENT_IMAGE.value is not None: config['container_specs']['kubernetes_deployment_startup'][ @@ -76,24 +124,60 @@ def Prepare(benchmark_spec: bm_spec.BenchmarkSpec): def Run(benchmark_spec: bm_spec.BenchmarkSpec) -> List[sample.Sample]: """Runs the benchmark and collects the results. + Collects three categories of metrics: + 1. max_pod_ready_time — existing metric, preserved unchanged. + 2. per_pod_ready_time — new: one Sample per pod with pod name in metadata. + 3. cpu_utilization_* — new: peak/mean/count via background collector. + + All samples carry scenario/workload/cloud/replicas metadata. + Args: benchmark_spec: The benchmark specification. Raises: - RuntimeError: Raised if no pods are ready after the deployment has finished - rolling out. + RuntimeError: Raised if no pods are ready after the deployment rolls out. Returns: A list of sample.Sample objects. """ image = benchmark_spec.container_specs['kubernetes_deployment_startup'].image - kubernetes_commands.ApplyManifest( - DEPLOYMENT_YAML.value, - name='startup', - image=image, - ) - kubernetes_commands.WaitForRollout('deployment/startup', timeout=600) + # ── Base metadata attached to every sample ──────────────────────────────── + base_metadata: Dict[str, Any] = { + 'scenario': SCENARIO.value, + 'workload': WORKLOAD.value, + 'cloud': FLAGS.cloud, + } + + # ── Start CPU background collector ──────────────────────────────────────── + all_samples: List[sample.Sample] = [] + stop = threading.Event() + cpu_collector = _CpuUtilizationCollector(all_samples, stop) + + # Apply manifest and wait for rollout inside a try/finally so we always + # stop the collector even if WaitForRollout raises. + try: + kubernetes_commands.ApplyManifest( + DEPLOYMENT_YAML.value, + name='startup', + image=image, + ) + + # Run CPU collector in parallel with the rollout wait, exactly like + # KubernetesMetricsCollector in kubernetes_hpa_benchmark.py. + collector_thread = threading.Thread( + target=cpu_collector.ObserveCpuUtilization, + daemon=True, + ) + collector_thread.start() + + kubernetes_commands.WaitForRollout('deployment/startup', timeout=600) + + finally: + stop.set() + collector_thread.join(timeout=_CPU_POLL_INTERVAL_SECS * 3) + + # ── Parse pod conditions (existing logic, unchanged) ────────────────────── pod_name_to_start_end_times: dict[str, tuple[int, int]] = ( collections.defaultdict(lambda: (0, 0)) ) @@ -111,14 +195,49 @@ def Run(benchmark_spec: bm_spec.BenchmarkSpec) -> List[sample.Sample]: c.epoch_time, ) + if not pod_name_to_start_end_times: + raise RuntimeError('No pods became ready') + + # ── Metric 1: max_pod_ready_time (existing, unchanged) ─────────────────── max_pod_ready_t = -1 for _, times in pod_name_to_start_end_times.items(): t = times[1] - times[0] max_pod_ready_t = max(max_pod_ready_t, t) - if max_pod_ready_t > -1: - return [sample.Sample('max_pod_ready_time', max_pod_ready_t, 'seconds', {})] - raise RuntimeError('No pods became ready') + if max_pod_ready_t < 0: + raise RuntimeError('No pods became ready') + + all_samples.append( + sample.Sample( + 'max_pod_ready_time', + max_pod_ready_t, + 'seconds', + {**base_metadata}, + ) + ) + + # ── Metric 2: per_pod_ready_time (new — PR 1) ──────────────────────────── + # Emit one Sample per pod so callers can compute p50/p90 across replicas. + for pod_name, (start_t, end_t) in pod_name_to_start_end_times.items(): + pod_ready_t = end_t - start_t + if pod_ready_t >= 0: + all_samples.append( + sample.Sample( + 'per_pod_ready_time', + pod_ready_t, + 'seconds', + {**base_metadata, 'pod_name': pod_name}, + ) + ) + + logging.info( + '[startup] max_pod_ready_time=%.2fs across %d pod(s)', + max_pod_ready_t, + len(pod_name_to_start_end_times), + ) + + # CPU samples were appended to all_samples by the collector thread. + return all_samples def Cleanup(benchmark_spec: bm_spec.BenchmarkSpec): @@ -128,3 +247,166 @@ def Cleanup(benchmark_spec: bm_spec.BenchmarkSpec): benchmark_spec: The benchmark specification. """ del benchmark_spec + + +# --------------------------------------------------------------------------- +# CPU Utilization Background Collector (new — PR 1) +# --------------------------------------------------------------------------- + + +class _CpuUtilizationCollector: + """Polls CPU utilization in a background thread during the startup window. + + Follows the KubernetesMetricsCollector / _Observe pattern from + kubernetes_hpa_benchmark.py exactly: + - _Observe(fn) loops calling fn() and appending results to self._samples. + - Stops when self._stop is signalled. + - Ignores IssueCommandError / IssueCommandTimeoutError (gaps in data OK). + + Emits three samples on completion: + cpu_utilization_peak_millicores — maximum reading during startup window. + cpu_utilization_mean_millicores — mean across all polls. + cpu_utilization_reading_count — number of successful polls. + """ + + def __init__( + self, + samples: List[sample.Sample], + stop: threading.Event, + ): + """Initialises the collector. + + Args: + samples: Shared sample list. CPU samples are appended here when + ObserveCpuUtilization() finishes. + stop: Threading event. Collector loops until this is set. + """ + self._samples = samples + self._stop = stop + self._readings: List[float] = [] + self._lock = threading.Lock() + + def ObserveCpuUtilization(self) -> None: + """Polls CPU millicores until stop is set; appends aggregate samples. + + Intended to be run in a background thread alongside WaitForRollout(). + Matches the ObserveNumReplicas / ObserveNumNodes pattern in + kubernetes_hpa_benchmark.py. + """ + self._Observe(self._PollCpuMillicoresSample) + + # Emit aggregate samples after the loop ends. + with self._lock: + readings = list(self._readings) + + if not readings: + logging.warning('[startup/cpu] No CPU readings collected.') + return + + peak = max(readings) + mean = sum(readings) / len(readings) + count = len(readings) + + logging.info( + '[startup/cpu] peak=%.1f mean=%.1f count=%d millicores', + peak, mean, count, + ) + + self._samples.extend([ + sample.Sample( + 'cpu_utilization_peak_millicores', peak, 'millicores', {} + ), + sample.Sample( + 'cpu_utilization_mean_millicores', mean, 'millicores', {} + ), + sample.Sample( + 'cpu_utilization_reading_count', count, 'count', {} + ), + ]) + + def _PollCpuMillicoresSample(self) -> List[sample.Sample]: + """Issues kubectl top pods and returns a transient sample list. + + The return value is a list so _Observe() can call self._samples.extend() + on it (matching the KubernetesMetricsCollector interface). The actual + reading is also stored in self._readings for aggregate computation. + + Returns: + A single-element list with the current CPU reading, or empty on error. + """ + cpu_m = _GetTotalCpuMillicores() + if cpu_m is None: + return [] + with self._lock: + self._readings.append(cpu_m) + # Return an empty list — we do NOT emit a per-poll sample (too noisy). + # Aggregates are emitted in ObserveCpuUtilization() after the loop. + return [] + + def _Observe( + self, + observe_fn: Callable[[], List[sample.Sample]], + ) -> None: + """Calls observe_fn in a loop until self._stop is set. + + Copied verbatim from KubernetesMetricsCollector._Observe() in + kubernetes_hpa_benchmark.py — same error handling, same 1 s wait. + + Args: + observe_fn: Function returning a list of samples to extend into + self._samples. + """ + success_count = 0 + failure_count = 0 + while True: + try: + self._samples.extend(observe_fn()) + success_count += 1 + except ( + errors.VmUtil.IssueCommandError, + errors.VmUtil.IssueCommandTimeoutError, + ) as e: + logging.warning( + '[startup/cpu] Ignoring poll error (gap in data): %s', e + ) + failure_count += 1 + + if self._stop.wait(timeout=_CPU_POLL_INTERVAL_SECS): + logging.info( + '[startup/cpu] Stopping after %d successes / %d failures', + success_count, failure_count, + ) + return + + +def _GetTotalCpuMillicores() -> float | None: + """Returns total CPU millicores across all pods via kubectl top pods. + + Returns: + Total CPU millicores, or None if the command fails or output is empty. + """ + try: + stdout, _, rc = kubectl.RunKubectlCommand( + ['top', 'pods', '--no-headers'], + raise_on_failure=False, + ) + if rc != 0 or not stdout.strip(): + return None + + total_m = 0.0 + for line in stdout.strip().splitlines(): + parts = line.split() + # kubectl top format: NAME CPU(cores) MEMORY(bytes) + if len(parts) < 2: + continue + cpu_str = parts[1] + if cpu_str.endswith('m'): + total_m += float(cpu_str[:-1]) + else: + # Expressed as fractional cores (e.g. "1" = 1000m). + total_m += float(cpu_str) * 1000.0 + + return total_m + except (ValueError, IndexError) as e: + logging.debug('[startup/cpu] Parse error: %s', e) + return None diff --git a/tests/linux_benchmarks/kubernetes_deployment_startup_benchmark_test.py b/tests/linux_benchmarks/kubernetes_deployment_startup_benchmark_test.py index 728622a4dd..3769f76582 100644 --- a/tests/linux_benchmarks/kubernetes_deployment_startup_benchmark_test.py +++ b/tests/linux_benchmarks/kubernetes_deployment_startup_benchmark_test.py @@ -11,80 +11,307 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +"""Tests for kubernetes_deployment_startup_benchmark (PR 1).""" +import threading import unittest from unittest import mock -from perfkitbenchmarker import benchmark_spec -from perfkitbenchmarker import sample -from perfkitbenchmarker.linux_benchmarks import kubernetes_deployment_startup_benchmark as kdsb -from perfkitbenchmarker.resources.container_service import kubernetes_commands -from perfkitbenchmarker.resources.container_service import kubernetes_conditions +from absl.testing import flagsaver +from perfkitbenchmarker import errors +from perfkitbenchmarker.linux_benchmarks import ( + kubernetes_deployment_startup_benchmark as bench, +) from tests import pkb_common_test_case -class KubernetesDeploymentStartupBenchmarkTest( - pkb_common_test_case.PkbCommonTestCase -): - """Tests for the kubernetes_deployment_startup_benchmark.""" - - def setUp(self): - super().setUp() - self.spec = mock.Mock(spec=benchmark_spec.BenchmarkSpec) - self.spec.container_cluster = mock.Mock() - self.spec.container_specs = { - 'kubernetes_deployment_startup': mock.Mock(image='test_image') - } - - @mock.patch.object(kubernetes_commands, 'WaitForRollout') - @mock.patch.object(kubernetes_commands, 'ApplyManifest') - @mock.patch.object( - kubernetes_conditions, 'GetStatusConditionsForResourceType' - ) - def testRun( - self, mock_get_conditions, mock_apply_manifest, mock_wait_for_rollout - ): - """Tests the Run method with mock pod data.""" - mock_get_conditions.return_value = [ - mock.Mock( - event='PodReadyToStartContainers', - resource_name='pod1', - epoch_time=10, - ), - mock.Mock(event='Ready', resource_name='pod1', epoch_time=20), - mock.Mock( - event='PodReadyToStartContainers', - resource_name='pod2', - epoch_time=12, - ), - mock.Mock(event='Ready', resource_name='pod2', epoch_time=25), +def _MakeCondition(resource_name, event, epoch_time): + """Returns a mock KubernetesStatusCondition.""" + c = mock.MagicMock() + c.resource_name = resource_name + c.event = event + c.epoch_time = epoch_time + return c + + +def _MakeSpec(image='test_image'): + """Returns a mock BenchmarkSpec.""" + bm = mock.MagicMock() + bm.container_specs = { + 'kubernetes_deployment_startup': mock.MagicMock(image=image) + } + return bm + + +def _DefaultConditions(): + return [ + _MakeCondition('pod-0', 'PodReadyToStartContainers', 1000), + _MakeCondition('pod-0', 'Ready', 1030), + ] + + +def _RunWithConditions(conditions, flag_kwargs=None): + """Runs bench.Run() with mocked kubectl calls.""" + flag_kwargs = flag_kwargs or {'cloud': 'GCP'} + with mock.patch.object( + bench.kubernetes_commands, 'ApplyManifest' + ), mock.patch.object( + bench.kubernetes_commands, 'WaitForRollout' + ), mock.patch.object( + bench.kubernetes_conditions, + 'GetStatusConditionsForResourceType', + return_value=conditions, + ), mock.patch.object( + bench, '_GetTotalCpuMillicores', return_value=None + ), flagsaver.flagsaver(**flag_kwargs): + return bench.Run(_MakeSpec()) + + +# --------------------------------------------------------------------------- +# Existing metric: max_pod_ready_time (preserved from original) +# --------------------------------------------------------------------------- + + +class MaxPodReadyTimeTest(pkb_common_test_case.PkbCommonTestCase): + """Tests for max_pod_ready_time metric (existing, preserved).""" + + def testEmitsMaxPodReadyTime(self): + """max_pod_ready_time is always emitted.""" + samples = _RunWithConditions(_DefaultConditions()) + self.assertIn('max_pod_ready_time', {s.metric for s in samples}) + + def testMaxPodReadyTimeValue(self): + """max_pod_ready_time equals worst pod across all replicas.""" + conditions = [ + _MakeCondition('pod-0', 'PodReadyToStartContainers', 1000), + _MakeCondition('pod-0', 'Ready', 1020), + _MakeCondition('pod-1', 'PodReadyToStartContainers', 1000), + _MakeCondition('pod-1', 'Ready', 1035), ] - result = kdsb.Run(self.spec) + samples = _RunWithConditions(conditions) + by_metric = {s.metric: s.value for s in samples} + self.assertAlmostEqual(by_metric['max_pod_ready_time'], 35) - mock_apply_manifest.assert_called_with( - kdsb.DEPLOYMENT_YAML.value, name='startup', image='test_image' - ) - mock_wait_for_rollout.assert_called_with('deployment/startup', timeout=600) - self.assertLen(result, 1) - self.assertEqual( - result[0], - sample.Sample( - 'max_pod_ready_time', 13, 'seconds', {}, result[0].timestamp - ), + def testOriginalTwoPodsValue(self): + """Preserves original test: 2 pods with times 10 and 13; max=13.""" + conditions = [ + _MakeCondition('pod1', 'PodReadyToStartContainers', 10), + _MakeCondition('pod1', 'Ready', 20), + _MakeCondition('pod2', 'PodReadyToStartContainers', 12), + _MakeCondition('pod2', 'Ready', 25), + ] + samples = _RunWithConditions(conditions) + by_metric = {s.metric: s.value for s in samples} + self.assertAlmostEqual(by_metric['max_pod_ready_time'], 13) + + def testApplyManifestAndWaitForRolloutCalled(self): + """Verify ApplyManifest and WaitForRollout are called correctly.""" + with mock.patch.object( + bench.kubernetes_commands, 'ApplyManifest' + ) as mock_apply, mock.patch.object( + bench.kubernetes_commands, 'WaitForRollout' + ) as mock_wait, mock.patch.object( + bench.kubernetes_conditions, + 'GetStatusConditionsForResourceType', + return_value=_DefaultConditions(), + ), mock.patch.object( + bench, '_GetTotalCpuMillicores', return_value=None + ), flagsaver.flagsaver(cloud='GCP'): + bench.Run(_MakeSpec()) + mock_apply.assert_called_with( + bench.DEPLOYMENT_YAML.value, name='startup', image='test_image' ) + mock_wait.assert_called_with('deployment/startup', timeout=600) - @mock.patch.object(kubernetes_commands, 'WaitForRollout') - @mock.patch.object(kubernetes_commands, 'ApplyManifest') - @mock.patch.object( - kubernetes_conditions, 'GetStatusConditionsForResourceType' - ) - def testRunNoPods( - self, mock_get_conditions, mock_apply_manifest, mock_wait_for_rollout - ): - """Tests the Run method when no pods are found.""" - mock_get_conditions.return_value = [] + def testRaisesWhenNoPodsReady(self): + """RuntimeError raised when no pod conditions found.""" with self.assertRaises(RuntimeError): - kdsb.Run(self.spec) + _RunWithConditions([]) + + +# --------------------------------------------------------------------------- +# New metric: per_pod_ready_time (PR 1) +# --------------------------------------------------------------------------- + + +class PerPodReadyTimeTest(pkb_common_test_case.PkbCommonTestCase): + """Tests for per_pod_ready_time metric (PR 1).""" + + def testEmitsOnePerPod(self): + """One per_pod_ready_time sample per pod.""" + conditions = [ + _MakeCondition('pod-0', 'PodReadyToStartContainers', 1000), + _MakeCondition('pod-0', 'Ready', 1025), + _MakeCondition('pod-1', 'PodReadyToStartContainers', 1000), + _MakeCondition('pod-1', 'Ready', 1040), + ] + samples = _RunWithConditions(conditions) + per_pod = [s for s in samples if s.metric == 'per_pod_ready_time'] + self.assertLen(per_pod, 2) + + def testPerPodCarriesPodName(self): + """per_pod_ready_time metadata contains pod_name.""" + conditions = [ + _MakeCondition('pod-abc', 'PodReadyToStartContainers', 1000), + _MakeCondition('pod-abc', 'Ready', 1030), + ] + samples = _RunWithConditions(conditions) + per_pod = [s for s in samples if s.metric == 'per_pod_ready_time'] + self.assertEqual(per_pod[0].metadata['pod_name'], 'pod-abc') + + def testPerPodValue(self): + """per_pod_ready_time value equals end_time - start_time.""" + conditions = [ + _MakeCondition('pod-x', 'PodReadyToStartContainers', 2000), + _MakeCondition('pod-x', 'Ready', 2045), + ] + samples = _RunWithConditions(conditions) + per_pod = [s for s in samples if s.metric == 'per_pod_ready_time'] + self.assertAlmostEqual(per_pod[0].value, 45) + + +# --------------------------------------------------------------------------- +# Sample metadata (PR 1) +# --------------------------------------------------------------------------- + + +class SampleMetadataTest(pkb_common_test_case.PkbCommonTestCase): + """Tests for scenario/workload/cloud metadata on samples (PR 1).""" + + def testAllPodSamplesCarryMetadata(self): + """scenario, workload, cloud present on all pod samples.""" + samples = _RunWithConditions( + _DefaultConditions(), + flag_kwargs={ + 'cloud': 'GCP', + 'kubernetes_deployment_startup_scenario': 'baseline', + 'kubernetes_deployment_startup_workload': 'jvm', + }, + ) + pod_samples = [ + s for s in samples + if s.metric in ('max_pod_ready_time', 'per_pod_ready_time') + ] + for s in pod_samples: + self.assertEqual(s.metadata['scenario'], 'baseline') + self.assertEqual(s.metadata['workload'], 'jvm') + self.assertEqual(s.metadata['cloud'], 'GCP') + + +# --------------------------------------------------------------------------- +# CPU utilization collector (PR 1) +# --------------------------------------------------------------------------- + + +class CpuUtilizationCollectorTest(pkb_common_test_case.PkbCommonTestCase): + """Tests for _CpuUtilizationCollector (PR 1).""" + + def _MakeCollector(self): + samples = [] + stop = threading.Event() + collector = bench._CpuUtilizationCollector(samples, stop) + return collector, samples, stop + + def testEmitsPeakMeanCount(self): + """ObserveCpuUtilization emits peak/mean/count samples.""" + collector, samples, stop = self._MakeCollector() + collector._readings = [100.0, 200.0, 300.0] + stop.set() + collector.ObserveCpuUtilization() + metrics = {s.metric for s in samples} + self.assertIn('cpu_utilization_peak_millicores', metrics) + self.assertIn('cpu_utilization_mean_millicores', metrics) + self.assertIn('cpu_utilization_reading_count', metrics) + + def testPeakAndMeanValues(self): + """Peak = max, mean = arithmetic mean of readings.""" + collector, samples, stop = self._MakeCollector() + collector._readings = [100.0, 200.0, 300.0] + stop.set() + collector.ObserveCpuUtilization() + by_metric = {s.metric: s.value for s in samples} + self.assertAlmostEqual(by_metric['cpu_utilization_peak_millicores'], 300.0) + self.assertAlmostEqual(by_metric['cpu_utilization_mean_millicores'], 200.0) + self.assertEqual(by_metric['cpu_utilization_reading_count'], 3) + + def testNoSamplesWhenNoReadings(self): + """No CPU samples emitted when no readings collected.""" + collector, samples, stop = self._MakeCollector() + collector._readings = [] + stop.set() + collector.ObserveCpuUtilization() + self.assertEqual(samples, []) + + def testObserveIgnoresIssueCommandError(self): + """_Observe continues on IssueCommandError.""" + collector, _, stop = self._MakeCollector() + call_count = [0] + + def flaky(): + call_count[0] += 1 + if call_count[0] < 3: + raise errors.VmUtil.IssueCommandError('transient') + stop.set() + return [] + + collector._Observe(flaky) + self.assertEqual(call_count[0], 3) + + def testThreadSafety(self): + """Concurrent appends to _readings are thread-safe.""" + collector, _, _ = self._MakeCollector() + errs = [] + + def append_readings(): + try: + for _ in range(50): + with collector._lock: + collector._readings.append(1.0) + except Exception as e: # pylint: disable=broad-except + errs.append(e) + + threads = [threading.Thread(target=append_readings) for _ in range(4)] + for t in threads: + t.start() + for t in threads: + t.join() + self.assertEqual(errs, []) + self.assertEqual(len(collector._readings), 200) + + +# --------------------------------------------------------------------------- +# _GetTotalCpuMillicores (PR 1) +# --------------------------------------------------------------------------- + + +class GetTotalCpuMillicoresTest(pkb_common_test_case.PkbCommonTestCase): + """Tests for _GetTotalCpuMillicores helper (PR 1).""" + + def _MockKubectl(self, stdout, rc=0): + return mock.patch.object( + bench.kubectl, 'RunKubectlCommand', + return_value=(stdout, '', rc), + ) + + def testParsesMiliSuffix(self): + with self._MockKubectl('pod-abc 250m 128Mi\n'): + self.assertAlmostEqual(bench._GetTotalCpuMillicores(), 250.0) + + def testParsesCoreSuffix(self): + with self._MockKubectl('pod-abc 1 512Mi\n'): + self.assertAlmostEqual(bench._GetTotalCpuMillicores(), 1000.0) + + def testSumsMultiplePods(self): + with self._MockKubectl('pod-0 100m 64Mi\npod-1 150m 64Mi\n'): + self.assertAlmostEqual(bench._GetTotalCpuMillicores(), 250.0) + + def testReturnsNoneOnError(self): + with self._MockKubectl('', rc=1): + self.assertIsNone(bench._GetTotalCpuMillicores()) + + def testReturnsNoneOnEmpty(self): + with self._MockKubectl(''): + self.assertIsNone(bench._GetTotalCpuMillicores()) if __name__ == '__main__':