rptest: move cluster config export to @cluster

This should be done only once, don't do it on service stop, but in the
@cluster cleanup code, before we stop redpanda.

This saves six seconds on every test, since the second export (due to
the second stop) took six seconds. Hat tip to Stephan for noticing
this.

Author: travisdowns

tests/rptest/services/cluster.py

@@ -154,6 +154,8 @@ def _do_post_test_checks(
                         redpanda.cloud_storage_diagnostics()
                     raise
 
+            redpanda.export_cluster_config()
+
             if isinstance(redpanda, RedpandaService):
                 if test_failed:
                     redpanda.cloud_storage_diagnostics()

tests/rptest/services/redpanda.py

@@ -1212,13 +1212,13 @@ class has both implementation and abstract methods, only for methods which
 
     def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
-        # Now ensure that this base is part of the right type of object.
-        # For these checks to work, this __init__ method should appear in
-        # the MRO after all the __init__ methods that would set these properties
-        self._check_attr("context", TestContext)
-        self._check_attr("logger", Logger)
         self._usage_stats = UsageStats()
 
+    @property
+    @abstractmethod
+    def logger(self) -> Logger:
+        pass
+
     @property
     def usage_stats(self) -> UsageStats:
         """
@@ -1248,6 +1248,9 @@ def kafka_client_security(self) -> KafkaClientSecurity:
         on this broker."""
         pass
 
+    def export_cluster_config(self) -> None:
+        self.logger.debug("export_cluster_config not implemented for this service")
+
     def wait_until(
         self,
         fn: Callable[[], Any],
@@ -1331,16 +1334,6 @@ def wrapped():
             logger=logger,
         )
 
-    def _check_attr(self, name: str, t: Type) -> None:
-        v = getattr(self, name, None)
-        mro = self.__class__.__mro__
-        assert v is not None, (
-            f"RedpandaServiceABC was missing attribute {name} after __init__: {mro}\n"
-        )
-        assert isinstance(v, t), (
-            f"{name} had wrong type, expected {t} but was {type(v)}"
-        )
-
     def _extract_samples(
         self, metrics: Generator[Metric, Any, None], sample_pattern: str, node: Any
     ) -> list[MetricSample]:
@@ -1582,7 +1575,6 @@ def __init__(
         # we save the test context under both names since RedpandaService and Service
         # save them under these two names, respetively
         self.context = self._context = context
-        self.logger = context.logger
 
         super().__init__()
 
@@ -1667,6 +1659,10 @@ def __init__(
             f"Not enough brokers: test needs {self._min_brokers} but cluster has {node_count}"
         )
 
+    @property
+    def logger(self) -> Logger:
+        return self._context.logger
+
     @property
     def kubectl(self) -> KubectlTool:
         assert self.__kubectl, "kubectl accessed before cluster was started?"
@@ -2379,7 +2375,7 @@ def copy_from_pod(params):
         return {}
 
 
-class RedpandaService(RedpandaServiceABC, Service):
+class RedpandaService(Service, RedpandaServiceABC):
     PERSISTENT_ROOT = "/var/lib/redpanda"
     TRIM_LOGS_KEY = "trim_logs"
     DATA_DIR = os.path.join(PERSISTENT_ROOT, "data")
@@ -4449,11 +4445,11 @@ def get_version_if_not_head(self, node):
             return self.get_version(node)
         return None
 
-    def stop(self, **kwargs: Any) -> None:
+    def export_cluster_config(self) -> None:
         """
-        Override default stop() to execude stop_node in parallel
+        Export the cluster configuration of all nodes to the ducktape log
+        directory for later inspection.
         """
-        self._stop_time = time.time()  # The last time stop is invoked
         self.logger.info("%s: exporting cluster config" % self.who_am_i())
 
         service_dir = os.path.join(
@@ -4477,6 +4473,12 @@ def stop(self, **kwargs: Any) -> None:
             # will not be able to get it from the admin API
             self.logger.info(f"{self.who_am_i()}: error getting config: {e}")
 
+    def stop(self, **kwargs: Any) -> None:
+        """
+        Override default stop() to execude stop_node in parallel
+        """
+        self._stop_time = time.time()  # The last time stop is invoked
+
         self.logger.info("%s: stopping service" % self.who_am_i())
 
         self.for_nodes(self.nodes, lambda n: self.stop_node(n, **kwargs))