rpk connect: guard major version upgrades

Now it will prompt the user if the upgrade
contains a major version bump.

We will also print the installed version
in `rpk connect install` for better UX.

src/go/rpk/pkg/cli/connect/connect.go

@@ -69,7 +69,7 @@ func NewCommand(fs afero.Fs, p *config.Params, execFn func(string, []string) err
 					return
 				}
 				fmt.Println("Downloading latest Redpanda Connect")
-				path, err := installConnect(cmd.Context(), fs, "latest")
+				path, _, err := installConnect(cmd.Context(), fs, "latest")
 				out.MaybeDie(err, "unable to install Redpanda Connect: %v; if running on an air-gapped environment you may install 'redpanda-connect' with your package manager.", err)
 				pluginPath = path
 			}

src/go/rpk/pkg/cli/connect/install.go

@@ -53,10 +53,10 @@ You may force the installation of Redpanda Connect using the --force flag.
 				}
 				out.Exit("Redpanda connect is already installed.\nIf you want to upgrade to the latest version, please run 'rpk connect upgrade'.")
 			}
-			_, err = installConnect(cmd.Context(), fs, version)
+			_, installedVersion, err := installConnect(cmd.Context(), fs, version)
 			out.MaybeDie(err, "unable to install Redpanda Connect: %v; if running on an air-gapped environment you may install 'redpanda-connect' with your package manager.", err)
 
-			fmt.Println("Redpanda Connect successfully installed.")
+			fmt.Printf("Redpanda Connect %v successfully installed.\n", installedVersion)
 		},
 	}
 	cmd.Flags().BoolVar(&force, "force", false, "Force install of Redpanda Connect")
@@ -69,20 +69,20 @@ You may force the installation of Redpanda Connect using the --force flag.
 // the latest plugin. Version string let you select a specific version to
 // download, if "latest" or an empty string is passed, it will download the
 // latest version.
-func installConnect(ctx context.Context, fs afero.Fs, version string) (path string, err error) {
+func installConnect(ctx context.Context, fs afero.Fs, version string) (path, installedVersion string, err error) {
 	// We check this before calling the API, to avoid getting the binary
 	// and later fail due to a user setting.
 	pluginDir, err := plugin.DefaultBinPath()
 	if err != nil {
-		return "", fmt.Errorf("unable to determine plugin default path: %v", err)
+		return "", "", fmt.Errorf("unable to determine plugin default path: %v", err)
 	}
 
-	art, _, err := getConnectArtifact(ctx, version)
+	art, ver, err := getConnectArtifact(ctx, version)
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
-
-	return downloadAndInstallConnect(ctx, fs, pluginDir, art.Path, art.Sha256)
+	path, err = downloadAndInstallConnect(ctx, fs, pluginDir, art.Path, art.Sha256)
+	return path, ver, err
 }
 
 func getConnectArtifact(ctx context.Context, version string) (connectArtifact, string, error) {

src/go/rpk/pkg/cli/connect/upgrade.go

@@ -10,17 +10,22 @@
 package connect
 
 import (
+	"context"
 	"fmt"
+	"os/exec"
 	"strings"
 
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/redpanda"
+
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/out"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/plugin"
 	"github.com/spf13/afero"
 	"github.com/spf13/cobra"
 )
 
 func upgradeCommand(fs afero.Fs) *cobra.Command {
-	return &cobra.Command{
+	var noConfirm bool
+	cmd := &cobra.Command{
 		Use:   "upgrade",
 		Short: "Upgrade to the latest Redpanda Connect version",
 		Args:  cobra.NoArgs,
@@ -47,10 +52,54 @@ func upgradeCommand(fs afero.Fs) *cobra.Command {
 			if strings.HasPrefix(currentSha, art.Sha256) {
 				out.Exit("Redpanda Connect already up-to-date")
 			}
+			currentVersion, err := connectVersion(cmd.Context(), connect.Path)
+			out.MaybeDie(err, "unable to determine current version of Redpanda Connect: %v", err)
+
+			if !noConfirm {
+				latestVersion, err := redpanda.VersionFromString(version)
+				out.MaybeDie(err, "unable to parse latest version of Redpanda Connect: %v", err)
+				if latestVersion.Major > currentVersion.Major {
+					confirmed, err := out.Confirm("Confirm major version upgrade from %v to %v?", currentVersion.String(), latestVersion.String())
+					out.MaybeDie(err, "unable to confirm upgrade: %v", err)
+					if !confirmed {
+						out.Exit("Upgrade canceled.")
+					}
+				}
+			}
+
 			_, err = downloadAndInstallConnect(cmd.Context(), fs, pluginDir, art.Path, art.Sha256)
 			out.MaybeDieErr(err)
 
-			fmt.Printf("Redpanda Connect successfully upgraded to the latest version (%v).\n", version)
+			fmt.Printf("Redpanda Connect successfully upgraded from %v to the latest version (%v).\n", currentVersion.String(), version)
 		},
 	}
+	cmd.Flags().BoolVar(&noConfirm, "no-confirm", false, "Disable confirmation prompt for major version upgrades")
+	return cmd
+}
+
+// connectVersion executes rpk connect --version and parses the current version
+// from the output.
+func connectVersion(ctx context.Context, connectPath string) (redpanda.Version, error) {
+	versionCmd := exec.CommandContext(ctx, connectPath, "--version")
+	var sb strings.Builder
+	versionCmd.Stdout = &sb
+	if err := versionCmd.Run(); err != nil {
+		return redpanda.Version{}, err
+	}
+	// Command output is:
+	//   Version: <Version>
+	//   Date: <Build date>
+	versionPrefix := "Version: "
+	var versionStr string
+	for _, l := range strings.Split(sb.String(), "\n") {
+		if strings.HasPrefix(l, versionPrefix) {
+			versionStr = strings.TrimPrefix(l, versionPrefix)
+			break
+		}
+	}
+	version, err := redpanda.VersionFromString(strings.TrimSpace(versionStr))
+	if err != nil {
+		return redpanda.Version{}, fmt.Errorf("unable to determine Redpanda version from %q: %v", versionStr, err)
+	}
+	return version, nil
 }

tests/rptest/clients/rpk.py

@@ -2002,7 +2002,7 @@ def uninstall_connect(self):
         return self._run_connect(cmd)
 
     def upgrade_connect(self):
-        cmd = ["upgrade"]
+        cmd = ["upgrade", "--no-confirm"]
         return self._run_connect(cmd)
 
     def connect_version(self):