rptest: limit node operation parallelism in cloud tests

It has been observed in tests that creating too many parallel ssh
sessions to the agent node can leave the teleport service on it in an
unrecoverable state.

Even in cases where the agent instance type is changed to something
larger(i.e, m5.large) the k8s API seems to timeout requests if too many
are outstanding in parallel.

Author: ballard26

tests/rptest/services/redpanda.py

@@ -1275,6 +1275,7 @@ class has both implementation and abstract methods, only for methods which
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
         self._usage_stats = UsageStats()
+        self._max_workers: int | None = None
 
     @property
     @abstractmethod
@@ -1405,10 +1406,9 @@ def wrapped():
         )
 
     def for_nodes(self, nodes: Collection[U], cb: Callable[[U], T]) -> list[T]:
-        n_workers = len(nodes)
-        if n_workers > 0:
+        if len(nodes) > 0:
             with concurrent.futures.ThreadPoolExecutor(
-                max_workers=n_workers
+                max_workers=self._max_workers
             ) as executor:
                 # The list() wrapper is to cause futures to be evaluated here+now
                 # (including throwing any exceptions) and not just spawned in background.
@@ -1797,6 +1797,10 @@ def __init__(
 
         super().__init__()
 
+        # Cloudv2 agents run on very small instances that can easily be
+        # overwhelmed by too many concurrent ssh sessions.
+        self._max_workers = 10
+
         self.config_profile_name = config_profile_name
         self._min_brokers = min_brokers
         self._superuser = RedpandaService.SUPERUSER_CREDENTIALS

tests/rptest/services/utils.py

@@ -151,6 +151,7 @@ def __init__(
         self._raise_on_errors: bool = self._context.globals.get(
             self.RAISE_ON_ERRORS_KEY, True
         )
+        self._max_workers: int | None = None
 
         # Prepare matching terms
         self.match_terms: list[str] = list(self.DEFAULT_MATCH_TERMS)
@@ -242,7 +243,7 @@ def scan_one(
         bad_lines: NodeToLines = {}
 
         # Run scans in parallel
-        with ThreadPoolExecutor() as executor:
+        with ThreadPoolExecutor(max_workers=self._max_workers) as executor:
             futures = [executor.submit(scan_one, v, n) for v, n in versioned_nodes]
             for fut in as_completed(futures):
                 node, vl = fut.result()
@@ -358,6 +359,9 @@ def __init__(
         # Prepare capture functions
         self.kubectl = kubectl
         self.test_start_time = test_start_time
+        # Cloudv2 agents run on very small instances that can easily be
+        # overwhelmed by too many concurrent ssh sessions.
+        self._max_workers = 10
 
     def _capture_log(self, node: CloudBroker, expr: str) -> Generator[str, None, None]:
         """Capture log and check test timing.