rptest: move cluster config export to @cluster

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))