From 0db3d1dd75ec06ac7c9ce2ce20dc5d76c89db3c2 Mon Sep 17 00:00:00 2001
From: r-vasquez <rvasque3@gmail.com>
Date: Fri, 1 Nov 2024 10:09:06 -0700
Subject: [PATCH] rpk: add `rpk debug remote bundle`

Fixes DEVEX-44

This commit introduces the rpk debug remote bundle
command, which allows the user to request a debug
bundle using the Admin API.
---
 MODULE.bazel                                  |   1 +
 src/go/rpk/go.mod                             |   2 +-
 src/go/rpk/pkg/cli/debug/BUILD                |   1 +
 src/go/rpk/pkg/cli/debug/bundle/bundle.go     |   2 +-
 src/go/rpk/pkg/cli/debug/common/BUILD         |   9 +
 src/go/rpk/pkg/cli/debug/common/common.go     |  18 +-
 src/go/rpk/pkg/cli/debug/debug.go             |   2 +
 src/go/rpk/pkg/cli/debug/remotebundle/BUILD   |  27 +++
 .../rpk/pkg/cli/debug/remotebundle/cancel.go  | 121 ++++++++++
 .../pkg/cli/debug/remotebundle/download.go    | 216 +++++++++++++++++
 .../rpk/pkg/cli/debug/remotebundle/remote.go  |  67 ++++++
 .../rpk/pkg/cli/debug/remotebundle/start.go   | 217 ++++++++++++++++++
 .../rpk/pkg/cli/debug/remotebundle/status.go  | 148 ++++++++++++
 13 files changed, 820 insertions(+), 11 deletions(-)
 create mode 100644 src/go/rpk/pkg/cli/debug/common/BUILD
 create mode 100644 src/go/rpk/pkg/cli/debug/remotebundle/BUILD
 create mode 100644 src/go/rpk/pkg/cli/debug/remotebundle/cancel.go
 create mode 100644 src/go/rpk/pkg/cli/debug/remotebundle/download.go
 create mode 100644 src/go/rpk/pkg/cli/debug/remotebundle/remote.go
 create mode 100644 src/go/rpk/pkg/cli/debug/remotebundle/start.go
 create mode 100644 src/go/rpk/pkg/cli/debug/remotebundle/status.go

diff --git a/MODULE.bazel b/MODULE.bazel
index 30196e7f31172..6f17614fc24e8 100644
--- a/MODULE.bazel
+++ b/MODULE.bazel
@@ -185,6 +185,7 @@ use_repo(
     "com_github_docker_go_connections",
     "com_github_docker_go_units",
     "com_github_fatih_color",
+    "com_github_google_uuid",
     "com_github_hamba_avro",
     "com_github_hamba_avro_v2",
     "com_github_hashicorp_go_multierror",
diff --git a/src/go/rpk/go.mod b/src/go/rpk/go.mod
index 6eac1aa8d03cd..2382d2292a4c3 100644
--- a/src/go/rpk/go.mod
+++ b/src/go/rpk/go.mod
@@ -23,6 +23,7 @@ require (
 	github.com/docker/go-connections v0.5.0
 	github.com/docker/go-units v0.5.0
 	github.com/fatih/color v1.18.0
+	github.com/google/uuid v1.6.0
 	github.com/hamba/avro/v2 v2.27.0
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51
@@ -94,7 +95,6 @@ require (
 	github.com/google/gnostic-models v0.6.9-0.20230804172637-c7be7c783f49 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
-	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.1 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
diff --git a/src/go/rpk/pkg/cli/debug/BUILD b/src/go/rpk/pkg/cli/debug/BUILD
index f939904cb81f1..5626b2b1ca1f6 100644
--- a/src/go/rpk/pkg/cli/debug/BUILD
+++ b/src/go/rpk/pkg/cli/debug/BUILD
@@ -10,6 +10,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//src/go/rpk/pkg/cli/debug/bundle",
+        "//src/go/rpk/pkg/cli/debug/remotebundle",
         "//src/go/rpk/pkg/config",
         "@com_github_spf13_afero//:afero",
         "@com_github_spf13_cobra//:cobra",
diff --git a/src/go/rpk/pkg/cli/debug/bundle/bundle.go b/src/go/rpk/pkg/cli/debug/bundle/bundle.go
index 00b2ca7180b3b..922cbf70ca9bf 100644
--- a/src/go/rpk/pkg/cli/debug/bundle/bundle.go
+++ b/src/go/rpk/pkg/cli/debug/bundle/bundle.go
@@ -155,7 +155,7 @@ func NewCommand(fs afero.Fs, p *config.Params) *cobra.Command {
 
 	f := cmd.Flags()
 	f.StringVarP(&outFile, outputFlag, "o", "", "The file path where the debug file will be written (default ./<timestamp>-bundle.zip)")
-	f.DurationVar(&timeout, "timeout", 31*time.Second, "How long to wait for child commands to execute (e.g. 30s, 1.5m)")
+	f.DurationVar(&timeout, "timeout", 31*time.Second, "How long to wait for child commands to execute. For example: 30s, 1.5m")
 	f.StringVar(&uploadURL, "upload-url", "", "If provided, where to upload the bundle in addition to creating a copy on disk")
 	// Debug bundle options.
 	opts.InstallFlags(f)
diff --git a/src/go/rpk/pkg/cli/debug/common/BUILD b/src/go/rpk/pkg/cli/debug/common/BUILD
new file mode 100644
index 0000000000000..42a44d5843310
--- /dev/null
+++ b/src/go/rpk/pkg/cli/debug/common/BUILD
@@ -0,0 +1,9 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "common",
+    srcs = ["common.go"],
+    importpath = "github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/debug/common",
+    visibility = ["//visibility:public"],
+    deps = ["@com_github_spf13_pflag//:pflag"],
+)
diff --git a/src/go/rpk/pkg/cli/debug/common/common.go b/src/go/rpk/pkg/cli/debug/common/common.go
index 97d8113cea813..920b4f938530e 100644
--- a/src/go/rpk/pkg/cli/debug/common/common.go
+++ b/src/go/rpk/pkg/cli/debug/common/common.go
@@ -25,16 +25,16 @@ type DebugBundleSharedOptions struct {
 // InstallFlags installs the debug bundle flags that fills the debug bundle
 // options.
 func (o *DebugBundleSharedOptions) InstallFlags(f *pflag.FlagSet) {
-	f.StringVar(&o.ControllerLogsSizeLimit, "controller-logs-size-limit", "132MB", "The size limit of the controller logs that can be stored in the bundle (e.g. 3MB, 1GiB)")
-	f.DurationVar(&o.CPUProfilerWait, "cpu-profiler-wait", 30*time.Second, "For how long to collect samples for the CPU profiler (e.g. 30s, 1.5m). Must be higher than 15s")
-	f.StringVar(&o.LogsSizeLimit, "logs-size-limit", "100MiB", "Read the logs until the given size is reached (e.g. 3MB, 1GiB)")
-	f.StringVar(&o.LogsSince, "logs-since", "yesterday", "Include logs dated from specified date onward; (journalctl date format: YYYY-MM-DD, 'yesterday', or 'today'). Refer to journalctl documentation for more options")
-	f.StringVar(&o.LogsUntil, "logs-until", "", "Include logs older than the specified date; (journalctl date format: YYYY-MM-DD, 'yesterday', or 'today'). Refer to journalctl documentation for more options")
-	f.DurationVar(&o.MetricsInterval, "metrics-interval", 10*time.Second, "Interval between metrics snapshots (e.g. 30s, 1.5m)")
+	f.StringVar(&o.ControllerLogsSizeLimit, "controller-logs-size-limit", "132MB", "The size limit of the controller logs that can be stored in the bundle. For example: 3MB, 1GiB")
+	f.DurationVar(&o.CPUProfilerWait, "cpu-profiler-wait", 30*time.Second, "How long to collect samples for the CPU profiler. For example: 30s, 1.5m. Must be higher than 15s")
+	f.StringVar(&o.LogsSizeLimit, "logs-size-limit", "100MiB", "Read the logs until the given size is reached. For example: 3MB, 1GiB")
+	f.StringVar(&o.LogsSince, "logs-since", "yesterday", "Include logs dated from specified date onward; (journalctl date format: YYYY-MM-DD, 'yesterday', or 'today'). See the journalctl documentation for more options")
+	f.StringVar(&o.LogsUntil, "logs-until", "", "Include logs older than the specified date; (journalctl date format: YYYY-MM-DD, 'yesterday', or 'today'). See the journalctl documentation for more options")
+	f.DurationVar(&o.MetricsInterval, "metrics-interval", 10*time.Second, "Interval between metrics snapshots. For example: 30s, 1.5m")
 	f.IntVar(&o.MetricsSampleCount, "metrics-samples", 2, "Number of metrics samples to take (at the interval of --metrics-interval). Must be >= 2")
-	f.StringArrayVarP(&o.PartitionFlag, "partition", "p", nil, "Comma-separated partition IDs; when provided, rpk saves extra admin API requests for those partitions. Check help for extended usage")
-	f.StringVarP(&o.Namespace, "namespace", "n", "redpanda", "The namespace to use to collect the resources from (k8s only)")
-	f.StringArrayVarP(&o.LabelSelector, "label-selector", "l", []string{"app.kubernetes.io/name=redpanda"}, "Comma-separated label selectors to filter your resources. e.g: <label>=<value>,<label>=<value> (k8s only)")
+	f.StringArrayVarP(&o.PartitionFlag, "partition", "p", nil, "Comma-separated partition IDs. When provided, rpk saves extra Admin API requests for those partitions. See the help for extended usage")
+	f.StringVarP(&o.Namespace, "namespace", "n", "redpanda", "The namespace to use to collect the resources from (K8s only)")
+	f.StringArrayVarP(&o.LabelSelector, "label-selector", "l", []string{"app.kubernetes.io/name=redpanda"}, "Comma-separated label selectors to filter your resources. For example: <label>=<value>,<label>=<value> (K8s only)")
 }
 
 // SanitizeName replace any of the following characters with "-": "<", ">", ":",
diff --git a/src/go/rpk/pkg/cli/debug/debug.go b/src/go/rpk/pkg/cli/debug/debug.go
index a3c551edd4810..e3630d246908e 100644
--- a/src/go/rpk/pkg/cli/debug/debug.go
+++ b/src/go/rpk/pkg/cli/debug/debug.go
@@ -11,6 +11,7 @@ package debug
 
 import (
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/debug/bundle"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/debug/remotebundle"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/config"
 	"github.com/spf13/afero"
 	"github.com/spf13/cobra"
@@ -25,6 +26,7 @@ func NewCommand(fs afero.Fs, p *config.Params) *cobra.Command {
 	cmd.AddCommand(
 		bundle.NewCommand(fs, p),
 		NewInfoCommand(),
+		remotebundle.NewCommand(fs, p),
 	)
 
 	return cmd
diff --git a/src/go/rpk/pkg/cli/debug/remotebundle/BUILD b/src/go/rpk/pkg/cli/debug/remotebundle/BUILD
new file mode 100644
index 0000000000000..44687beeddc2f
--- /dev/null
+++ b/src/go/rpk/pkg/cli/debug/remotebundle/BUILD
@@ -0,0 +1,27 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "remotebundle",
+    srcs = [
+        "cancel.go",
+        "download.go",
+        "remote.go",
+        "start.go",
+        "status.go",
+    ],
+    importpath = "github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/debug/remotebundle",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//src/go/rpk/pkg/adminapi",
+        "//src/go/rpk/pkg/cli/debug/common",
+        "//src/go/rpk/pkg/config",
+        "//src/go/rpk/pkg/out",
+        "@com_github_docker_go_units//:go-units",
+        "@com_github_google_uuid//:uuid",
+        "@com_github_redpanda_data_common_go_rpadmin//:rpadmin",
+        "@com_github_spf13_afero//:afero",
+        "@com_github_spf13_cobra//:cobra",
+        "@io_k8s_apimachinery//pkg/labels",
+        "@org_golang_x_sync//errgroup",
+    ],
+)
diff --git a/src/go/rpk/pkg/cli/debug/remotebundle/cancel.go b/src/go/rpk/pkg/cli/debug/remotebundle/cancel.go
new file mode 100644
index 0000000000000..5dfbef95445e0
--- /dev/null
+++ b/src/go/rpk/pkg/cli/debug/remotebundle/cancel.go
@@ -0,0 +1,121 @@
+// Copyright 2024 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+package remotebundle
+
+import (
+	"fmt"
+	"os"
+	"sync"
+
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/config"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/out"
+	"github.com/spf13/afero"
+	"github.com/spf13/cobra"
+)
+
+type cancelResponse struct {
+	Broker   string
+	Canceled bool
+	Error    string
+}
+
+func newCancelCommand(fs afero.Fs, p *config.Params) *cobra.Command {
+	var noConfirm bool
+	var jobID string
+	cmd := &cobra.Command{
+		Use:   "cancel",
+		Short: "Cancel a remote bundle execution",
+		Long: `Cancel a remote bundle execution.
+
+This command cancels the debug collection process in a remote cluster that
+you configured in flags, environment variables, or your rpk profile.
+
+Use the flag '--job-id' to only cancel the debug bundle with
+the given job ID.
+
+Use the flag '--no-confirm' to avoid the confirmation prompt.
+`,
+		Args: cobra.NoArgs,
+		Run: func(cmd *cobra.Command, _ []string) {
+			p, err := p.LoadVirtualProfile(fs)
+			out.MaybeDie(err, "rpk unable to load config: %v", err)
+			config.CheckExitCloudAdmin(p)
+
+			status, anyErr, _, cache := executeBundleStatus(cmd.Context(), fs, p)
+			if jobID != "" {
+				status = filterStatusByJobID(status, jobID)
+			}
+			if !noConfirm {
+				printTextBundleStatus(status, anyErr)
+				confirmed, err := out.Confirm("Confirm debug bundle cancel from these brokers?")
+				out.MaybeDie(err, "unable to confirm cancel: %v;  if you want to select a single broker, use -X admin.hosts=<brokers,to,cancel>", err)
+				if !confirmed {
+					out.Exit("operation canceled; if you want to select a single broker, use -X admin.hosts=<brokers,to,cancel>")
+				}
+			}
+			var (
+				wg           sync.WaitGroup
+				rwMu         sync.RWMutex // read from cache.
+				mu           sync.Mutex   // write to status.
+				response     []cancelResponse
+				anyCancelErr bool
+			)
+			updateStatus := func(resp cancelResponse, err error) {
+				mu.Lock()
+				defer mu.Unlock()
+				anyCancelErr = anyCancelErr || err != nil
+				response = append(response, resp)
+			}
+			for _, s := range status {
+				wg.Add(1)
+				go func(addr, jobID string) {
+					defer wg.Done()
+					rwMu.RLock()
+					cl := cache[addr]
+					rwMu.RUnlock()
+					resp := cancelResponse{Broker: addr}
+					err := cl.CancelDebugBundleProcess(cmd.Context(), jobID)
+					if err != nil {
+						resp.Error = fmt.Sprintf("unable to cancel debug bundle: %s", tryMessageFromErr(err))
+						updateStatus(resp, err)
+						return
+					}
+					resp.Canceled = true
+					updateStatus(resp, nil)
+				}(s.Broker, s.JobID)
+			}
+			wg.Wait()
+			headers := []string{"broker", "canceled"}
+			if anyCancelErr {
+				headers = append(headers, "error")
+				defer os.Exit(1)
+			}
+
+			tw := out.NewTable(headers...)
+			defer tw.Flush()
+			for _, row := range response {
+				tw.PrintStructFields(row)
+			}
+		},
+	}
+	cmd.Flags().StringVar(&jobID, "job-id", "", "ID of the job to cancel the debug bundle")
+	cmd.Flags().BoolVar(&noConfirm, "no-confirm", false, "Disable confirmation prompt")
+	return cmd
+}
+
+func filterStatusByJobID(status []statusResponse, jobID string) []statusResponse {
+	var filtered []statusResponse
+	for _, s := range status {
+		if s.JobID == jobID {
+			filtered = append(filtered, s)
+		}
+	}
+	return filtered
+}
diff --git a/src/go/rpk/pkg/cli/debug/remotebundle/download.go b/src/go/rpk/pkg/cli/debug/remotebundle/download.go
new file mode 100644
index 0000000000000..9dd8f9c598757
--- /dev/null
+++ b/src/go/rpk/pkg/cli/debug/remotebundle/download.go
@@ -0,0 +1,216 @@
+// Copyright 2024 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+package remotebundle
+
+import (
+	"archive/zip"
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/redpanda-data/common-go/rpadmin"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/debug/common"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/config"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/out"
+	"github.com/spf13/afero"
+	"github.com/spf13/cobra"
+	"golang.org/x/sync/errgroup"
+)
+
+type downloadResponse struct {
+	Broker     string
+	Downloaded bool
+	Error      string
+}
+
+func newDownloadCommand(fs afero.Fs, p *config.Params) *cobra.Command {
+	var (
+		noConfirm bool
+		jobID     string
+		outFile   string
+	)
+	cmd := &cobra.Command{
+		Use:   "download",
+		Short: "Download a remote debug bundle",
+		Long: `Download a remote debug bundle.
+
+This command downloads the debug collection process in a remote cluster that
+you configured in flags, environment variables, or your rpk profile.
+
+Use the flag '--job-id' to only download the debug bundle with
+the given job ID.
+
+Use the flag '--no-confirm' to avoid the confirmation prompt.
+`,
+		Args: cobra.NoArgs,
+		Run: func(cmd *cobra.Command, _ []string) {
+			p, err := p.LoadVirtualProfile(fs)
+			out.MaybeDie(err, "rpk unable to load config: %v", err)
+			config.CheckExitCloudAdmin(p)
+
+			status, _, _, cache := executeBundleStatus(cmd.Context(), fs, p)
+			if jobID != "" {
+				status = filterStatusByJobID(status, jobID)
+			}
+			ready := filterReadyBrokers(status)
+			if len(ready) == 0 {
+				out.Exit("There are no bundles ready for download; to check status run 'rpk debug remote-bundle status'")
+			}
+			if len(ready) != len(status) {
+				fmt.Printf("WARNING: There are (%d) bundles not ready to download; to check the status run 'rpk debug remote-bundle status'\n", len(status)-len(ready))
+			}
+
+			if !noConfirm {
+				printTextBundleStatus(ready, false)
+				confirmed, err := out.Confirm("Confirm debug bundle download from these brokers?")
+				out.MaybeDie(err, "unable to confirm download: %v;  if you want to select a single broker, use -X admin.hosts=<brokers,to,cancel>", err)
+				if !confirmed {
+					out.Exit("operation canceled; if you want to select a single broker, use -X admin.hosts=<brokers,to,cancel>")
+				}
+			}
+			downloadPath, err := fileLocation(fs, outFile)
+			out.MaybeDie(err, "unable to determine download path: %v", err)
+
+			anyResponseErr, response, err := downloadBundle(cmd.Context(), fs, downloadPath, status, cache)
+			out.MaybeDie(err, "unable to download bundle: %v", err)
+
+			headers := []string{"broker", "downloaded"}
+			if anyResponseErr {
+				headers = append(headers, "error")
+				defer os.Exit(1)
+			}
+			tw := out.NewTable(headers...)
+			for _, row := range response {
+				tw.PrintStructFields(row)
+			}
+			tw.Flush()
+
+			if anyResponseErr {
+				fmt.Printf("\nDownloaded remote debug bundle to %v, but there were errors while downloading the broker files.\n", downloadPath)
+				os.Exit(1)
+			}
+			fmt.Printf("\nSuccessfully downloaded remote debug bundle to %v\n", downloadPath)
+		},
+	}
+	cmd.Flags().StringVarP(&outFile, "output", "o", "", "The file path where the debug file will be written (default ./<timestamp>-remote-bundle.zip)")
+	cmd.Flags().StringVar(&jobID, "job-id", "", "ID of the job to download the debug bundle from")
+	cmd.Flags().BoolVar(&noConfirm, "no-confirm", false, "Disable confirmation prompt")
+	return cmd
+}
+
+func downloadBundle(ctx context.Context, fs afero.Fs, downloadPath string, status []statusResponse, cache map[string]*rpadmin.AdminAPI) (bool, []downloadResponse, error) {
+	var (
+		rwMu     sync.RWMutex // read from cache
+		statusMu sync.Mutex   // write to status
+		zipMu    sync.Mutex   // write to file
+		response []downloadResponse
+		anyErr   bool
+	)
+	f, err := fs.OpenFile(downloadPath, os.O_CREATE|os.O_WRONLY, os.FileMode(0o755)) // This is the final zip file, a zip of zips.
+	if err != nil {
+		return false, nil, fmt.Errorf("unable to create zip file: %v", err)
+	}
+	w := zip.NewWriter(f)
+	defer w.Close()
+
+	downloadRoot := strings.TrimSuffix(filepath.Base(downloadPath), ".zip")
+	updateStatus := func(resp downloadResponse, err error) {
+		statusMu.Lock()
+		defer statusMu.Unlock()
+		anyErr = anyErr || err != nil
+		response = append(response, resp)
+	}
+	writeToZip := func(broker, jobID string, httpResp *http.Response) error {
+		if httpResp != nil {
+			zipMu.Lock()
+			defer zipMu.Unlock()
+			wr, err := w.CreateHeader(&zip.FileHeader{
+				// Each individual zip file will be stored under a directory
+				// with the broker address.
+				Name:     filepath.Join(downloadRoot, common.SanitizeName(broker), jobID+".zip"),
+				Method:   zip.Deflate,
+				Modified: time.Now(),
+			})
+			if err != nil {
+				return err
+			}
+			_, err = io.Copy(wr, httpResp.Body)
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+
+	grp, gCtx := errgroup.WithContext(ctx)
+	for _, s := range status {
+		grp.Go(func() error {
+			rwMu.RLock()
+			cl := cache[s.Broker]
+			rwMu.RUnlock()
+			resp := downloadResponse{Broker: s.Broker}
+			httpResp, err := cl.DownloadDebugBundleFile(gCtx, s.filename)
+			if err != nil {
+				resp.Error = fmt.Sprintf("unable to cancel debug bundle: %s", tryMessageFromErr(err))
+				updateStatus(resp, err)
+				return nil // No need to return error, we print to this to the table.
+			}
+			defer httpResp.Body.Close()
+			resp.Downloaded = true
+			updateStatus(resp, nil)
+			err = writeToZip(resp.Broker, s.JobID, httpResp)
+			if err != nil {
+				return fmt.Errorf("error writing debug bundle of %v: %v", s.Broker, err)
+			}
+			return nil
+		})
+	}
+	if err := grp.Wait(); err != nil {
+		return false, nil, err
+	}
+	return anyErr, response, nil
+}
+
+func filterReadyBrokers(status []statusResponse) []statusResponse {
+	var filtered []statusResponse
+	for _, s := range status {
+		if strings.EqualFold(s.Status, "success") {
+			filtered = append(filtered, s)
+		}
+	}
+	return filtered
+}
+
+func fileLocation(fs afero.Fs, path string) (string, error) {
+	// If it's empty, use "./<timestamp>-remote-bundle.zip"
+	if path == "" {
+		path = fmt.Sprintf("%d-remote-bundle.zip", time.Now().Unix())
+	} else if isDir, _ := afero.IsDir(fs, path); isDir {
+		return "", fmt.Errorf("output file path is a directory, please specify the name of the file")
+	}
+
+	var finalPath string
+	// Check for file extension, if extension is empty, defaults to .zip
+	switch ext := filepath.Ext(path); ext {
+	case ".zip":
+		finalPath = path
+	case "":
+		finalPath = path + ".zip"
+	default:
+		return "", fmt.Errorf("extension %q not supported", ext)
+	}
+	return finalPath, nil
+}
diff --git a/src/go/rpk/pkg/cli/debug/remotebundle/remote.go b/src/go/rpk/pkg/cli/debug/remotebundle/remote.go
new file mode 100644
index 0000000000000..6e2b9f6897fb9
--- /dev/null
+++ b/src/go/rpk/pkg/cli/debug/remotebundle/remote.go
@@ -0,0 +1,67 @@
+// Copyright 2024 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+package remotebundle
+
+import (
+	"errors"
+	"strings"
+
+	"github.com/redpanda-data/common-go/rpadmin"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/config"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/out"
+	"github.com/spf13/afero"
+	"github.com/spf13/cobra"
+)
+
+func NewCommand(fs afero.Fs, p *config.Params) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "remote-bundle",
+		Short: "Create a debug bundle from a remote cluster",
+		Long: `Create a debug bundle from a remote cluster.
+
+The remote-bundle commands can be used to get a debug bundle of your remote 
+cluster configured in flags, environment variables, or your rpk profile.
+`,
+		Args: cobra.NoArgs,
+	}
+	cmd.AddCommand(
+		newStartCommand(fs, p),
+		newStatusCommand(fs, p),
+		newDownloadCommand(fs, p),
+		newCancelCommand(fs, p),
+	)
+	return cmd
+}
+
+func printBrokers(addresses []string) {
+	tw := out.NewTable("broker")
+	defer tw.Flush()
+	for _, address := range addresses {
+		tw.Print(address)
+	}
+}
+
+func tryMessageFromErr(err error) string {
+	if he := (*rpadmin.HTTPResponseError)(nil); errors.As(err, &he) {
+		if body, err := he.DecodeGenericErrorBody(); err == nil {
+			return body.Message
+		}
+	}
+	return strings.TrimSpace(err.Error())
+}
+
+func isAlreadyStartedErr(err error) bool {
+	if he := (*rpadmin.HTTPResponseError)(nil); errors.As(err, &he) {
+		if body, err := he.DecodeGenericErrorBody(); err == nil {
+			return body.Code == rpadmin.DebugBundleErrorCodeProcessAlreadyRunning
+		}
+	}
+	return false
+}
diff --git a/src/go/rpk/pkg/cli/debug/remotebundle/start.go b/src/go/rpk/pkg/cli/debug/remotebundle/start.go
new file mode 100644
index 0000000000000..ba477a28956ca
--- /dev/null
+++ b/src/go/rpk/pkg/cli/debug/remotebundle/start.go
@@ -0,0 +1,217 @@
+// Copyright 2024 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+package remotebundle
+
+import (
+	"fmt"
+	"os"
+	"strings"
+	"sync"
+
+	"github.com/docker/go-units"
+	"github.com/google/uuid"
+	"github.com/redpanda-data/common-go/rpadmin"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/adminapi"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/debug/common"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/config"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/out"
+	"github.com/spf13/afero"
+	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/labels"
+)
+
+type startResponse struct {
+	Broker string
+	JobID  string
+	Error  string
+}
+
+func newStartCommand(fs afero.Fs, p *config.Params) *cobra.Command {
+	var (
+		noConfirm bool
+		jobID     string
+		opts      remoteBundleOptions
+	)
+	cmd := &cobra.Command{
+		Use:   "start",
+		Short: "Start a remote debug bundle collection in your cluster",
+		Long: `Start a remote debug bundle collection in your cluster.
+
+This command starts the debug collection process in a remote cluster that
+you configured in flags, environment variables, or your rpk profile.
+
+After starting the debug collection process, you can query the status with
+'rpk debug remote-bundle status'. When it completes, you can download it with
+'rpk debug remote-bundle download'.
+
+Use the flag '--no-confirm' to avoid the confirmation prompt.
+`,
+		Args: cobra.NoArgs,
+		PreRun: func(_ *cobra.Command, _ []string) {
+			clsl, err := units.FromHumanSize(opts.ControllerLogsSizeLimit)
+			out.MaybeDie(err, "unable to parse --controller-logs-size-limit: %v", err)
+			opts.controllerLogsSizeLimitBytes = int32(clsl)
+
+			lsl, err := units.FromHumanSize(opts.LogsSizeLimit)
+			out.MaybeDie(err, "unable to parse --logs-size-limit: %v", err)
+			opts.logsSizeLimitBytes = int32(lsl)
+
+			if len(opts.labelSelectorMap) > 0 {
+				labelsMap, err := labels.ConvertSelectorToLabelsMap(strings.Join(opts.LabelSelector, ","))
+				out.MaybeDie(err, "unable to parse label-selector flag: %v", err)
+				opts.labelSelectorMap = labelsMap
+			}
+		},
+		Run: func(cmd *cobra.Command, _ []string) {
+			p, err := p.LoadVirtualProfile(fs)
+			out.MaybeDie(err, "rpk unable to load config: %v", err)
+			config.CheckExitCloudAdmin(p)
+
+			if !noConfirm {
+				printBrokers(p.AdminAPI.Addresses)
+				confirmed, err := out.Confirm("Confirm debug bundle collection from these brokers?")
+				out.MaybeDie(err, "unable to confirm collection: %v;  if you want to select a single broker, use -X admin.hosts=<brokers,to,collect>", err)
+				if !confirmed {
+					out.Exit("operation canceled; if you want to select a single broker, use -X admin.hosts=<brokers,to,collect>")
+				}
+			}
+			if jobID == "" {
+				id, err := uuid.NewRandom()
+				out.MaybeDie(err, "unable to generate a UUID for the Job ID: %v", err)
+				jobID = id.String()
+			}
+			var (
+				response       []startResponse
+				wg             sync.WaitGroup
+				mu             sync.Mutex
+				anyErr, anyOK  bool
+				alreadyRunning bool
+			)
+			updateStatus := func(resp startResponse, err error) {
+				mu.Lock()
+				defer mu.Unlock()
+				anyErr = anyErr || err != nil
+				anyOK = anyOK || err == nil
+				alreadyRunning = alreadyRunning || isAlreadyStartedErr(err)
+				response = append(response, resp)
+			}
+			for _, addr := range p.AdminAPI.Addresses {
+				wg.Add(1)
+				go func(addr string) {
+					defer wg.Done()
+
+					resp := startResponse{Broker: addr}
+					client, err := adminapi.NewHostClient(fs, p, addr)
+					if err != nil {
+						resp.Error = fmt.Sprintf("unable to connect: %s", tryMessageFromErr(err))
+						updateStatus(resp, err)
+						return
+					}
+					bundleResp, err := client.CreateDebugBundle(cmd.Context(), jobID, opts.toRpadminOptions(p)...)
+					if err != nil {
+						isAlreadyStartedErr(err)
+						resp.Error = fmt.Sprintf("unable to start debug bundle: %s", tryMessageFromErr(err))
+						updateStatus(resp, err)
+						return
+					}
+					resp.JobID = bundleResp.JobID
+					updateStatus(resp, nil)
+				}(addr)
+			}
+			wg.Wait()
+
+			headers := []string{"broker", "job-ID"}
+			if anyErr {
+				headers = append(headers, "error")
+				defer os.Exit(1)
+			}
+
+			tw := out.NewTable(headers...)
+			for _, row := range response {
+				tw.PrintStructFields(row)
+			}
+			tw.Flush()
+
+			if alreadyRunning {
+				fmt.Printf(`
+A debug bundle collection is already in process. To cancel it, run:
+  rpk debug remote-bundle cancel
+`)
+			}
+			if anyOK {
+				fmt.Printf(`
+The debug bundle collection process has started with Job-ID %v. To check the 
+status, run:
+  rpk debug remote-bundle status
+`, jobID)
+			}
+		},
+	}
+	f := cmd.Flags()
+	f.StringVar(&jobID, "job-id", "", "ID of the job to start debug bundle in")
+	f.BoolVar(&noConfirm, "no-confirm", false, "Disable confirmation prompt")
+	// Debug bundle options:
+	opts.InstallFlags(f)
+
+	return cmd
+}
+
+type remoteBundleOptions struct {
+	common.DebugBundleSharedOptions
+	controllerLogsSizeLimitBytes int32
+	logsSizeLimitBytes           int32
+	labelSelectorMap             map[string]string
+}
+
+func (o *remoteBundleOptions) toRpadminOptions(p *config.RpkProfile) []rpadmin.DebugBundleOption {
+	var opts []rpadmin.DebugBundleOption
+	if o.controllerLogsSizeLimitBytes > 0 {
+		opts = append(opts, rpadmin.WithControllerLogsSizeLimitBytes(o.controllerLogsSizeLimitBytes))
+	}
+	if cpuWait := int32(o.CPUProfilerWait.Seconds()); cpuWait > 0 {
+		opts = append(opts, rpadmin.WithCPUProfilerWaitSeconds(cpuWait))
+	}
+	if o.LogsSince != "" {
+		opts = append(opts, rpadmin.WithLogsSince(o.LogsSince))
+	}
+	if o.LogsUntil != "" {
+		opts = append(opts, rpadmin.WithLogsUntil(o.LogsUntil))
+	}
+	if o.logsSizeLimitBytes > 0 {
+		opts = append(opts, rpadmin.WithLogsSizeLimitBytes(o.logsSizeLimitBytes))
+	}
+	if mis := int32(o.MetricsInterval.Seconds()); mis > 0 {
+		opts = append(opts, rpadmin.WithMetricsIntervalSeconds(mis))
+	}
+	if o.MetricsSampleCount > 0 {
+		opts = append(opts, rpadmin.WithMetricsSamples(int32(o.MetricsSampleCount)))
+	}
+	if len(o.PartitionFlag) > 0 {
+		opts = append(opts, rpadmin.WithPartitions(o.PartitionFlag))
+	}
+	if o.Namespace != "" {
+		opts = append(opts, rpadmin.WithNamespace(o.Namespace))
+	}
+	if len(o.LabelSelector) > 0 {
+		dbls := make([]rpadmin.DebugBundleLabelSelector, 0, len(o.LabelSelector))
+		for k, v := range o.labelSelectorMap {
+			dbls = append(dbls, rpadmin.DebugBundleLabelSelector{Key: k, Value: v})
+		}
+		opts = append(opts, rpadmin.WithLabelSelector(dbls))
+	}
+	if p.HasSASLCredentials() {
+		s := p.KafkaAPI.SASL
+		opts = append(opts, rpadmin.WithSCRAMAuthentication(s.User, s.Password, s.Mechanism))
+	}
+	if tls := p.KafkaAPI.TLS; tls != nil {
+		opts = append(opts, rpadmin.WithTLS(true, tls.InsecureSkipVerify))
+	}
+	return opts
+}
diff --git a/src/go/rpk/pkg/cli/debug/remotebundle/status.go b/src/go/rpk/pkg/cli/debug/remotebundle/status.go
new file mode 100644
index 0000000000000..1a6bd60f64dc0
--- /dev/null
+++ b/src/go/rpk/pkg/cli/debug/remotebundle/status.go
@@ -0,0 +1,148 @@
+// Copyright 2024 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+package remotebundle
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"slices"
+	"sync"
+
+	"github.com/redpanda-data/common-go/rpadmin"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/adminapi"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/config"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/out"
+	"github.com/spf13/afero"
+	"github.com/spf13/cobra"
+)
+
+type statusResponse struct {
+	Broker string `json:"broker" yaml:"broker"`
+	Status string `json:"status,omitempty" yaml:"status,omitempty"`
+	JobID  string `json:"job_id,omitempty" yaml:"job_id,omitempty"`
+	Error  string `json:"error,omitempty" yaml:"error,omitempty"`
+
+	filename string
+}
+
+func newStatusCommand(fs afero.Fs, p *config.Params) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "status",
+		Short: "Get the status of the current debug bundle process",
+		Long: `Get the status of the current debug bundle process.
+
+This command prints the status of the debug bundle process in a remote cluster 
+that you configured in flags, environment variables, or your rpk profile.
+
+When the process completes, you can download it with 
+  'rpk debug remote-bundle download'.
+`,
+		Args: cobra.NoArgs,
+		Run: func(cmd *cobra.Command, _ []string) {
+			f := p.Formatter
+			if h, ok := f.Help([]statusResponse{}); ok {
+				out.Exit(h)
+			}
+			p, err := p.LoadVirtualProfile(fs)
+			out.MaybeDie(err, "rpk unable to load config: %v", err)
+			config.CheckExitCloudAdmin(p)
+
+			slices.Sort(p.AdminAPI.Addresses)
+			status, anyErr, anyOK, _ := executeBundleStatus(cmd.Context(), fs, p)
+
+			if isText, _, formatted, err := f.Format(status); !isText {
+				out.MaybeDie(err, "unable to print status in the required format %q: %v", f.Kind, err)
+				fmt.Println(formatted)
+				return
+			}
+			printTextBundleStatus(status, anyErr)
+
+			if anyOK {
+				fmt.Printf(`
+After the process is completed, you may retrieve the debug bundle using:
+  rpk debug remote-bundle download
+`)
+			}
+			if anyErr {
+				os.Exit(1)
+			}
+		},
+	}
+	p.InstallFormatFlag(cmd)
+	return cmd
+}
+
+func executeBundleStatus(ctx context.Context, fs afero.Fs, p *config.RpkProfile) (status []statusResponse, anyErr, anyOK bool, clientCache map[string]*rpadmin.AdminAPI) {
+	var (
+		wg       sync.WaitGroup
+		sMu, mMu sync.Mutex
+	)
+	clientCache = make(map[string]*rpadmin.AdminAPI)
+	updateStatus := func(resp statusResponse, isErr bool) {
+		sMu.Lock()
+		defer sMu.Unlock()
+		anyErr = anyErr || isErr
+		anyOK = anyOK || !isErr
+		status = append(status, resp)
+	}
+	addClientToCache := func(addr string, client *rpadmin.AdminAPI) {
+		mMu.Lock()
+		defer mMu.Unlock()
+		clientCache[addr] = client
+	}
+	for _, addr := range p.AdminAPI.Addresses {
+		wg.Add(1)
+		go func(addr string) {
+			defer wg.Done()
+
+			resp := statusResponse{
+				Broker: addr,
+			}
+			client, err := adminapi.NewHostClient(fs, p, addr)
+			if err != nil {
+				resp.Error = fmt.Sprintf("unable to connect: %s", tryMessageFromErr(err))
+				updateStatus(resp, true)
+				return
+			}
+			addClientToCache(addr, client)
+			bs, err := client.GetDebugBundleStatus(ctx)
+			if err != nil {
+				resp.Error = fmt.Sprintf("unable to get bundle status: %s", tryMessageFromErr(err))
+				updateStatus(resp, true)
+				return
+			}
+
+			resp.Status = bs.Status
+			resp.JobID = bs.JobID
+			resp.filename = bs.Filename
+			updateStatus(resp, false)
+		}(addr)
+	}
+	wg.Wait()
+	return
+}
+
+func printTextBundleStatus(status []statusResponse, anyErr bool) {
+	headers := []string{"broker", "status", "job-ID"}
+	if anyErr {
+		headers = append(headers, "error")
+	}
+	tw := out.NewTable(headers...)
+	defer tw.Flush()
+	for _, row := range status {
+		// Can't use tw.PrintStructField as we don't want to print filename.
+		if anyErr {
+			tw.Print(row.Broker, row.Status, row.JobID, row.Error)
+		} else {
+			tw.Print(row.Broker, row.Status, row.JobID)
+		}
+	}
+}