Merge pull request #23749 from vbotbuildovich/backport-pr-23393-v24.2…

….x-205

[v24.2.x] [CORE-7292] Fix `rpk connect` ducktape test in GCP

tests/rptest/clients/rpk.py

@@ -1982,47 +1982,3 @@ def run_mock_plugin(self, cmd):
         cmd = [self._rpk_binary(), "pluginmock"] + cmd
         out = self._execute(cmd)
         return json.loads(out)
-
-    def _run_connect(self, cmd, timeout=None):
-        cmd = [self._rpk_binary(), "connect"] + cmd
-        return self._execute(cmd, timeout=timeout)
-
-    def install_connect(self, version="", force=False):
-        cmd = ["install"]
-        if version != "":
-            cmd += ["--connect-version", version]
-
-        if force:
-            cmd += ["--force"]
-
-        # This command has a higher timeout than normal rpk
-        # commands as it downloads and install the RP Connect
-        # binary.
-        return self._run_connect(cmd, timeout=2 * DEFAULT_TIMEOUT)
-
-    def uninstall_connect(self):
-        cmd = ["uninstall"]
-
-        return self._run_connect(cmd)
-
-    def upgrade_connect(self):
-        cmd = ["upgrade", "--no-confirm"]
-        return self._run_connect(cmd)
-
-    def connect_version(self):
-        cmd = ["--version"]
-        out = self._run_connect(cmd)
-        """
-        Example output of rpk connect --version:
-        Version: 4.33.0
-        Date: 2024-08-13T21:40:38Z
-        """
-        pattern = r"Version:\s*([\d.]+)"
-        match = re.search(pattern, out)
-        if match:
-            return match.group(1)  # The version.
-        else:
-            raise RpkException(f"unable to parse connect version from {out}")
-
-    def run_connect_arbitrary(self, cmd):
-        return self._run_connect(cmd)

tests/rptest/clients/rpk_remote.py

@@ -7,6 +7,9 @@
 # the Business Source License, use of this software will be governed
 # by the Apache License, Version 2.0
 
+import re
+from rptest.clients.rpk import RpkException
+
 
 class RpkRemoteTool:
     """
@@ -173,3 +176,52 @@ def read_timestamps(self, topic: str, partition: int, count: int,
                 raise
 
             yield (int(offset), int(ts))
+
+    def plugin_list(self):
+        cmd = [self._rpk_binary(), "plugin", "list", "--local"]
+
+        return self._execute(cmd)
+
+    def _run_connect(self, cmd, timeout=None):
+        cmd = [self._rpk_binary(), "connect"] + cmd
+        return self._execute(cmd, timeout=timeout)
+
+    def install_connect(self, version="", force=False):
+        cmd = ["install"]
+        if version != "":
+            cmd += ["--connect-version", version]
+
+        if force:
+            cmd += ["--force"]
+
+        # This command has a higher timeout than normal rpk
+        # commands as it downloads and install the RP Connect
+        # binary.
+        return self._run_connect(cmd, timeout=60)
+
+    def uninstall_connect(self):
+        cmd = ["uninstall"]
+
+        return self._run_connect(cmd)
+
+    def upgrade_connect(self):
+        cmd = ["upgrade", "--no-confirm"]
+        return self._run_connect(cmd)
+
+    def connect_version(self):
+        cmd = ["--version"]
+        out = self._run_connect(cmd)
+        """
+        Example output of rpk connect --version:
+        Version: 4.33.0
+        Date: 2024-08-13T21:40:38Z
+        """
+        pattern = r"Version:\s*([\d.]+)"
+        match = re.search(pattern, out)
+        if match:
+            return match.group(1)  # The version.
+        else:
+            raise RpkException(f"unable to parse connect version from {out}")
+
+    def run_connect_arbitrary(self, cmd):
+        return self._run_connect(cmd)

tests/rptest/tests/rpk_connect_test.py

@@ -10,59 +10,83 @@
 from rptest.services.cluster import cluster
 from rptest.tests.redpanda_test import RedpandaTest
 from rptest.util import expect_exception
-from rptest.clients.rpk import RpkTool, RpkException
+from ducktape.utils.util import wait_until
+from rptest.clients.rpk_remote import RpkRemoteTool
+from rptest.clients.rpk import RpkException
 
 
 class RpkConnectTest(RedpandaTest):
     def __init__(self, ctx):
         super(RpkConnectTest, self).__init__(test_context=ctx)
         self._ctx = ctx
-        self._rpk = RpkTool(self.redpanda)
 
-    def cleanup_connect(self):
-        self._rpk.uninstall_connect()
-        assert "connect" not in self._rpk.plugin_list()
+    def cleanup_connect(self, node):
+        rpk_remote = RpkRemoteTool(self.redpanda, node)
+        rpk_remote.uninstall_connect()
+        assert "connect" not in rpk_remote.plugin_list()
+
+    def wait_until_installed(self, node):
+        def installed():
+            rpk_remote = RpkRemoteTool(self.redpanda, node)
+            plugin_list = rpk_remote.plugin_list()
+            return "connect" in plugin_list
+
+        wait_until(
+            installed,
+            timeout_sec=120,
+            backoff_sec=2,
+            err_msg=
+            f"could not find 'redpanda-connect' in plugin list after installing"
+        )
 
     @cluster(num_nodes=1)
     def test_manual_latest_install(self):
-        self.cleanup_connect()
-        self._rpk.install_connect()
-        assert "connect" in self._rpk.plugin_list()
+        node = self.redpanda.get_node(0)
+        self.cleanup_connect(node)
+
+        rpk_remote = RpkRemoteTool(self.redpanda, node)
+        rpk_remote.install_connect()
+        self.wait_until_installed(node)
 
     @cluster(num_nodes=1)
     def test_upgrade(self):
-        self.cleanup_connect()
+        node = self.redpanda.get_node(0)
+        self.cleanup_connect(node)
 
+        rpk_remote = RpkRemoteTool(self.redpanda, node)
         initial_version = "4.35.1"
-        self._rpk.install_connect(initial_version)
-        assert "connect" in self._rpk.plugin_list()
-        assert self._rpk.connect_version() == initial_version
+        rpk_remote.install_connect(initial_version)
+        self.wait_until_installed(node)
+        assert rpk_remote.connect_version() == initial_version
 
-        self._rpk.upgrade_connect()
+        rpk_remote.upgrade_connect()
+        self.wait_until_installed(node)
         # We assume here that is updated to latest.
-        assert self._rpk.connect_version() != initial_version
+        assert rpk_remote.connect_version() != initial_version
 
     @cluster(num_nodes=1)
     def test_automatic_install(self):
-        self.cleanup_connect()
+        node = self.redpanda.get_node(0)
+        self.cleanup_connect(node)
+        rpk_remote = RpkRemoteTool(self.redpanda, node)
 
         # rpk will not install if the user runs just --help, instead
         # it will display the help text and exit 0.
-        assert "Usage:" in self._rpk.run_connect_arbitrary(["--help"])
+        assert "Usage:" in rpk_remote.run_connect_arbitrary(["--help"])
 
         # rpk shouldn't install either if the user runs --version without
         # connect being installed first.
         with expect_exception(
                 RpkException,
                 lambda e: "rpk connect is not installed" in str(e)):
-            self._rpk.connect_version()
+            rpk_remote.connect_version()
 
         # Now, rpk should download and install on-the-fly Connect when
         # executing a subcommand.
-        assert "connect" not in self._rpk.plugin_list()
-        self._rpk.run_connect_arbitrary(["run", "--help"])
-        assert "connect" in self._rpk.plugin_list()
+        assert "connect" not in rpk_remote.plugin_list()
+        rpk_remote.run_connect_arbitrary(["run", "--help"])
+        self.wait_until_installed(node)
 
         # And executing --version now that is installed should return
         # something and exit 0.
-        assert self._rpk.connect_version() != ""
+        assert rpk_remote.connect_version() != ""