diff --git a/MODULE.bazel b/MODULE.bazel
index ad3dff060d94..1377d9cddc80 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 62668de279bd..70f6c177b64c 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/adminapi/admin.go b/src/go/rpk/pkg/adminapi/admin.go
index 8c67799a932e..2fa82f9ae41f 100644
--- a/src/go/rpk/pkg/adminapi/admin.go
+++ b/src/go/rpk/pkg/adminapi/admin.go
@@ -21,14 +21,12 @@ import (
 
 	"github.com/kr/text"
 	mTerm "github.com/moby/term"
-
-	"go.uber.org/zap"
-
 	"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/oauth"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/oauth/providers/auth0"
 	"github.com/spf13/afero"
+	"go.uber.org/zap"
 )
 
 // GenericErrorBody is the JSON decodable body that is produced by generic error
diff --git a/src/go/rpk/pkg/cli/debug/BUILD b/src/go/rpk/pkg/cli/debug/BUILD
index f939904cb81f..5626b2b1ca1f 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/BUILD b/src/go/rpk/pkg/cli/debug/bundle/BUILD
index 47f47f7a8eab..569dd675b2bd 100644
--- a/src/go/rpk/pkg/cli/debug/bundle/BUILD
+++ b/src/go/rpk/pkg/cli/debug/bundle/BUILD
@@ -11,6 +11,7 @@ go_library(
     importpath = "github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/debug/bundle",
     visibility = ["//visibility:public"],
     deps = [
+        "//src/go/rpk/pkg/cli/debug/common",
         "//src/go/rpk/pkg/config",
         "//src/go/rpk/pkg/httpapi",
         "//src/go/rpk/pkg/kafka",
diff --git a/src/go/rpk/pkg/cli/debug/bundle/bundle.go b/src/go/rpk/pkg/cli/debug/bundle/bundle.go
index d58d545d4825..922cbf70ca9b 100644
--- a/src/go/rpk/pkg/cli/debug/bundle/bundle.go
+++ b/src/go/rpk/pkg/cli/debug/bundle/bundle.go
@@ -18,6 +18,7 @@ import (
 	"time"
 
 	"github.com/docker/go-units"
+	"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/httpapi"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/kafka"
@@ -59,20 +60,8 @@ func NewCommand(fs afero.Fs, p *config.Params) *cobra.Command {
 	var (
 		outFile   string
 		uploadURL string
-
-		logsSince     string
-		logsUntil     string
-		logsSizeLimit string
-
-		controllerLogsSizeLimit string
-		namespace               string
-		labelSelector           []string
-		partitionFlag           []string
-
-		timeout            time.Duration
-		metricsInterval    time.Duration
-		metricsSampleCount int
-		cpuProfilerWait    time.Duration
+		timeout   time.Duration
+		opts      common.DebugBundleSharedOptions
 	)
 	cmd := &cobra.Command{
 		Use:   "bundle",
@@ -83,11 +72,11 @@ func NewCommand(fs afero.Fs, p *config.Params) *cobra.Command {
 			//  Redpanda queries for samples from Seastar every ~13 seconds by
 			//  default. Setting wait_ms to anything less than 13 seconds will
 			//  result in no samples being returned.
-			if cpuProfilerWait < 15*time.Second {
+			if opts.CPUProfilerWait < 15*time.Second {
 				out.Die("--cpu-profiler-wait must be higher than 15 seconds")
 			}
 
-			if metricsSampleCount < 2 {
+			if opts.MetricsSampleCount < 2 {
 				out.Die("--metrics-samples must be 2 or higher")
 			}
 
@@ -108,17 +97,17 @@ func NewCommand(fs afero.Fs, p *config.Params) *cobra.Command {
 			path, err := determineFilepath(fs, yActual, outFile, cmd.Flags().Changed(outputFlag))
 			out.MaybeDie(err, "unable to determine filepath %q: %v", outFile, err)
 
-			partitions, err := parsePartitionFlag(partitionFlag)
-			out.MaybeDie(err, "unable to parse partition flag %v: %v", partitionFlag, err)
+			partitions, err := parsePartitionFlag(opts.PartitionFlag)
+			out.MaybeDie(err, "unable to parse partition flag %v: %v", opts.PartitionFlag, err)
 
 			cl, err := kafka.NewFranzClient(fs, p)
 			out.MaybeDie(err, "unable to initialize kafka client: %v", err)
 			defer cl.Close()
 
-			logsLimit, err := units.FromHumanSize(logsSizeLimit)
+			logsLimit, err := units.FromHumanSize(opts.LogsSizeLimit)
 			out.MaybeDie(err, "unable to parse --logs-size-limit: %v", err)
 
-			controllerLogsLimit, err := units.FromHumanSize(controllerLogsSizeLimit)
+			controllerLogsLimit, err := units.FromHumanSize(opts.ControllerLogsSizeLimit)
 			out.MaybeDie(err, "unable to parse --controller-logs-size-limit: %v", err)
 			bp := bundleParams{
 				fs:                      fs,
@@ -126,20 +115,20 @@ func NewCommand(fs afero.Fs, p *config.Params) *cobra.Command {
 				y:                       y,
 				yActual:                 yActual,
 				cl:                      cl,
-				logsSince:               logsSince,
-				logsUntil:               logsUntil,
 				path:                    path,
-				namespace:               namespace,
 				logsLimitBytes:          int(logsLimit),
 				controllerLogLimitBytes: int(controllerLogsLimit),
 				timeout:                 timeout,
-				metricsInterval:         metricsInterval,
-				metricsSampleCount:      metricsSampleCount,
 				partitions:              partitions,
-				cpuProfilerWait:         cpuProfilerWait,
+				namespace:               opts.Namespace,
+				logsSince:               opts.LogsSince,
+				logsUntil:               opts.LogsUntil,
+				metricsInterval:         opts.MetricsInterval,
+				metricsSampleCount:      opts.MetricsSampleCount,
+				cpuProfilerWait:         opts.CPUProfilerWait,
 			}
-			if len(labelSelector) > 0 {
-				labelsMap, err := labels.ConvertSelectorToLabelsMap(strings.Join(labelSelector, ","))
+			if len(opts.LabelSelector) > 0 {
+				labelsMap, err := labels.ConvertSelectorToLabelsMap(strings.Join(opts.LabelSelector, ","))
 				out.MaybeDie(err, "unable to parse label-selector flag: %v", err)
 				bp.labelSelector = labelsMap
 			}
@@ -166,18 +155,10 @@ 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(&metricsInterval, "metrics-interval", 10*time.Second, "Interval between metrics snapshots (e.g. 30s, 1.5m)")
-	f.IntVar(&metricsSampleCount, "metrics-samples", 2, "Number of metrics samples to take (at the interval of --metrics-interval). Must be >= 2")
-	f.StringVar(&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(&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.StringVar(&logsSizeLimit, "logs-size-limit", "100MiB", "Read the logs until the given size is reached (e.g. 3MB, 1GiB)")
-	f.StringVar(&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(&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")
-	f.StringVarP(&namespace, "namespace", "n", "redpanda", "The namespace to use to collect the resources from (k8s only)")
-	f.StringArrayVarP(&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(&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.DurationVar(&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")
+	// Debug bundle options.
+	opts.InstallFlags(f)
 
 	return cmd
 }
diff --git a/src/go/rpk/pkg/cli/debug/bundle/bundle_k8s_linux.go b/src/go/rpk/pkg/cli/debug/bundle/bundle_k8s_linux.go
index 71a08f5090e2..b67d8a477a50 100644
--- a/src/go/rpk/pkg/cli/debug/bundle/bundle_k8s_linux.go
+++ b/src/go/rpk/pkg/cli/debug/bundle/bundle_k8s_linux.go
@@ -24,15 +24,14 @@ import (
 	"strings"
 	"time"
 
-	"github.com/redpanda-data/common-go/rpadmin"
-
-	authorizationv1 "k8s.io/api/authorization/v1"
-
 	"github.com/hashicorp/go-multierror"
+	"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/spf13/afero"
 	"go.uber.org/zap"
+	authorizationv1 "k8s.io/api/authorization/v1"
 	k8score "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -354,7 +353,7 @@ func saveSingleAdminAPICalls(ctx context.Context, ps *stepParams, fs afero.Fs, p
 				continue
 			}
 
-			aName := sanitizeName(a)
+			aName := common.SanitizeName(a)
 			r := []func() error{
 				func() error {
 					return requestAndSave(ctx, ps, fmt.Sprintf("admin/node_config_%v.json", aName), cl.RawNodeConfig)
@@ -427,7 +426,7 @@ func saveMetricsAPICalls(ctx context.Context, ps *stepParams, fs afero.Fs, p *co
 			}
 
 			endpoints := map[string]func(context.Context) ([]byte, error){"metrics": cl.PrometheusMetrics, "public_metrics": cl.PublicMetrics}
-			aName := sanitizeName(a)
+			aName := common.SanitizeName(a)
 			for endpointName, endpoint := range endpoints {
 				endpointPoller := func() error {
 					err := requestAndSave(ctx, ps, fmt.Sprintf("metrics/%v/t0_%s.txt", aName, endpointName), endpoint)
@@ -639,18 +638,6 @@ func parseJournalTime(str string, now time.Time) (time.Time, error) {
 	}
 }
 
-// sanitizeName replace any of the following characters with "-": "<", ">", ":",
-// `"`, "/", "|", "?", "*". This is to avoid having forbidden names in Windows
-// environments.
-func sanitizeName(name string) string {
-	forbidden := []string{"<", ">", ":", `"`, "/", `\`, "|", "?", "*"}
-	r := name
-	for _, s := range forbidden {
-		r = strings.Replace(r, s, "-", -1)
-	}
-	return r
-}
-
 func saveExtraFuncs(ctx context.Context, ps *stepParams, cl *rpadmin.AdminAPI, partitionFilters []topicPartitionFilter) (funcs []func() error) {
 	for _, tpf := range partitionFilters {
 		tpf := tpf
diff --git a/src/go/rpk/pkg/cli/debug/bundle/bundle_linux.go b/src/go/rpk/pkg/cli/debug/bundle/bundle_linux.go
index 016e68624d8c..e3ff95602f4b 100644
--- a/src/go/rpk/pkg/cli/debug/bundle/bundle_linux.go
+++ b/src/go/rpk/pkg/cli/debug/bundle/bundle_linux.go
@@ -36,6 +36,7 @@ import (
 	"github.com/beevik/ntp"
 	"github.com/docker/go-units"
 	"github.com/hashicorp/go-multierror"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/debug/common"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/config"
 	osutil "github.com/redpanda-data/redpanda/src/go/rpk/pkg/os"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/out"
@@ -61,7 +62,7 @@ func determineFilepath(fs afero.Fs, rp *config.RedpandaYaml, path string, isFlag
 	if path == "" {
 		timestamp := time.Now().Unix()
 		if rp.Redpanda.AdvertisedRPCAPI != nil {
-			path = fmt.Sprintf("%v-%d-bundle.zip", sanitizeName(rp.Redpanda.AdvertisedRPCAPI.Address), timestamp)
+			path = fmt.Sprintf("%v-%d-bundle.zip", common.SanitizeName(rp.Redpanda.AdvertisedRPCAPI.Address), timestamp)
 		} else {
 			path = fmt.Sprintf("%d-bundle.zip", timestamp)
 		}
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 000000000000..ca4298152d6c
--- /dev/null
+++ b/src/go/rpk/pkg/cli/debug/common/BUILD
@@ -0,0 +1,16 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+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"],
+)
+
+go_test(
+    name = "common_test",
+    srcs = ["common_test.go"],
+    embed = [":common"],
+    deps = ["@com_github_stretchr_testify//require"],
+)
diff --git a/src/go/rpk/pkg/cli/debug/common/common.go b/src/go/rpk/pkg/cli/debug/common/common.go
new file mode 100644
index 000000000000..920b4f938530
--- /dev/null
+++ b/src/go/rpk/pkg/cli/debug/common/common.go
@@ -0,0 +1,50 @@
+package common
+
+import (
+	"strings"
+	"time"
+
+	"github.com/spf13/pflag"
+)
+
+// DebugBundleSharedOptions are the options of a debug bundle that can be
+// shared between a normal and a remote bundle.
+type DebugBundleSharedOptions struct {
+	CPUProfilerWait         time.Duration
+	LogsSince               string
+	LogsUntil               string
+	MetricsInterval         time.Duration
+	MetricsSampleCount      int
+	PartitionFlag           []string
+	Namespace               string
+	LabelSelector           []string
+	LogsSizeLimit           string
+	ControllerLogsSizeLimit string
+}
+
+// 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. 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. 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 "-": "<", ">", ":",
+// `"`, "/", "|", "?", "*". This is to avoid having forbidden names in Windows
+// environments.
+func SanitizeName(name string) string {
+	forbidden := []string{"<", ">", ":", `"`, "/", `\`, "|", "?", "*"}
+	r := name
+	for _, s := range forbidden {
+		r = strings.Replace(r, s, "-", -1)
+	}
+	return r
+}
diff --git a/src/go/rpk/pkg/cli/debug/common/common_test.go b/src/go/rpk/pkg/cli/debug/common/common_test.go
new file mode 100644
index 000000000000..02fc88f06ae6
--- /dev/null
+++ b/src/go/rpk/pkg/cli/debug/common/common_test.go
@@ -0,0 +1,58 @@
+package common
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSanitizeName(t *testing.T) {
+	tests := []struct {
+		name  string
+		input string
+		exp   string
+	}{
+		{
+			name:  "No forbidden characters",
+			input: "validName",
+			exp:   "validName",
+		},
+		{
+			name:  "Single forbidden character",
+			input: "invalid:8083",
+			exp:   "invalid-8083",
+		},
+		{
+			name:  "Multiple forbidden characters",
+			input: "name/with|forbidden?chars",
+			exp:   "name-with-forbidden-chars",
+		},
+		{
+			name:  "Only forbidden characters",
+			input: `<>:\"/\\|?*`,
+			exp:   "-----------",
+		},
+		{
+			name:  "Empty string",
+			input: "",
+			exp:   "",
+		},
+		{
+			name:  "No change with already sanitized name",
+			input: "cleanName123",
+			exp:   "cleanName123",
+		},
+		{
+			name:  "Name with numbers and special characters",
+			input: "name123|test*<>",
+			exp:   "name123-test---",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			actual := SanitizeName(tt.input)
+			require.Equal(t, tt.exp, actual)
+		})
+	}
+}
diff --git a/src/go/rpk/pkg/cli/debug/debug.go b/src/go/rpk/pkg/cli/debug/debug.go
index a3c551edd481..e3630d246908 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 000000000000..44687beeddc2
--- /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 000000000000..5dfbef95445e
--- /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 000000000000..60b76a95d03b
--- /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.Die("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 000000000000..6e2b9f6897fb
--- /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 000000000000..ba477a28956c
--- /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 000000000000..1a6bd60f64dc
--- /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)
+		}
+	}
+}
diff --git a/tests/rptest/clients/rpk.py b/tests/rptest/clients/rpk.py
index 8d2d9fa9ff1b..bad49f3cf930 100644
--- a/tests/rptest/clients/rpk.py
+++ b/tests/rptest/clients/rpk.py
@@ -1993,3 +1993,38 @@ def run_mock_plugin(self, cmd):
         cmd = [self._rpk_binary(), "pluginmock"] + cmd
         out = self._execute(cmd)
         return json.loads(out)
+
+    def remote_bundle_start(self, extra_flags=[]):
+        cmd = ["start", "--no-confirm"]
+
+        if len(extra_flags) > 1:
+            cmd += extra_flags
+
+        return self._run_remote_bundle(cmd)
+
+    def remote_bundle_status(self, format="json"):
+        cmd = ["status", "--format", format]
+        out = self._run_remote_bundle(cmd)
+
+        return json.loads(out) if format == "json" else out
+
+    def remote_bundle_download(self, job_id, output_file=None):
+        cmd = ["download", "--no-confirm", "--job-id", job_id]
+
+        if output_file is not None:
+            cmd += ["--output", output_file]
+
+        return self._run_remote_bundle(cmd)
+
+    def remote_bundle_cancel(self, job_id):
+        cmd = ["cancel", "--no-confirm", "--job-id", job_id]
+        return self._run_remote_bundle(cmd)
+
+    def _run_remote_bundle(self, cmd, format=None):
+        cmd = [self._rpk_binary(), "debug", "remote-bundle"
+               ] + cmd + ["-X", "admin.hosts=" + self._admin_host()]
+
+        if format is not None:
+            cmd += ["--format", format]
+
+        return self._execute(cmd)
diff --git a/tests/rptest/tests/rpk_cluster_test.py b/tests/rptest/tests/rpk_cluster_test.py
index 2003c5668110..0522197cd276 100644
--- a/tests/rptest/tests/rpk_cluster_test.py
+++ b/tests/rptest/tests/rpk_cluster_test.py
@@ -22,7 +22,6 @@
 
 from rptest.tests.redpanda_test import RedpandaTest
 from rptest.clients.rpk import RpkTool, RpkException
-from rptest.clients.rpk_remote import RpkRemoteTool
 
 
 class RpkClusterTest(RedpandaTest):
@@ -53,82 +52,6 @@ def condition():
                    backoff_sec=1,
                    err_msg="No brokers found or output doesn't match")
 
-    @cluster(num_nodes=3)
-    def test_debug_bundle(self):
-        # The main RpkTool helper runs rpk on the test runner machine -- debug
-        # commands are run on redpanda nodes.
-        root_name = "bundle"
-        bundle_name = "bundle" + ".zip"
-        working_dir = "/tmp"
-        file_path = os.path.join(working_dir, bundle_name)
-        node = self.redpanda.nodes[0]
-
-        rpk_remote = RpkRemoteTool(self.redpanda, node)
-        output = rpk_remote.debug_bundle(file_path)
-        lines = output.split("\n")
-
-        # On error, rpk bundle returns 0 but writes error description to stdout
-        output_file = None
-        error_lines = []
-        any_errs = False
-        for l in lines:
-            self.logger.info(l)
-            if l.strip().startswith("* "):
-                error_lines.append(l)
-            elif 'errors occurred' in l:
-                any_errs = True
-            else:
-                m = re.match("^Debug bundle saved to '(.+)'$", l)
-                if m:
-                    output_file = m.group(1)
-
-        # Avoid false passes if our error line scraping gets broken
-        # by a format change.
-        if any_errs:
-            assert error_lines
-
-        filtered_errors = []
-        for l in error_lines:
-            if "dmidecode" in l:
-                # dmidecode doesn't work in ducktape containers, ignore
-                # errors about it.
-                continue
-            if re.match(r".* error querying .*\.ntp\..* i\/o timeout", l):
-                self.logger.error(f"Non-fatal transitory NTP error: {l}")
-            else:
-                self.logger.error(f"Bad output line: {l}")
-                filtered_errors.append(l)
-
-        assert not filtered_errors
-        assert output_file == file_path
-
-        node.account.copy_from(output_file, working_dir)
-
-        zf = zipfile.ZipFile(output_file)
-        files = zf.namelist()
-        assert f'{root_name}/redpanda.yaml' in files
-        assert f'{root_name}/redpanda.log' in files
-
-        # At least the first controller log is being saved:
-        assert f'{root_name}/controller-logs/redpanda/controller/0_0/0-1-v1.log' in files
-
-        # Cluster admin API calls:
-        assert f'{root_name}/admin/brokers.json' in files
-        assert f'{root_name}/admin/cluster_config.json' in files
-        assert f'{root_name}/admin/health_overview.json' in files
-
-        # Per-node admin API calls:
-        for n in self.redpanda.started_nodes():
-            # rpk will save 2 snapsots per metrics endpoint:
-            assert f'{root_name}/metrics/{n.account.hostname}-9644/t0_metrics.txt' in files
-            assert f'{root_name}/metrics/{n.account.hostname}-9644/t1_metrics.txt' in files
-            assert f'{root_name}/metrics/{n.account.hostname}-9644/t0_public_metrics.txt' in files
-            assert f'{root_name}/metrics/{n.account.hostname}-9644/t1_public_metrics.txt' in files
-            # and 1 cluster_view and node_config per node:
-            assert f'{root_name}/admin/cluster_view_{n.account.hostname}-9644.json' in files
-            assert f'{root_name}/admin/node_config_{n.account.hostname}-9644.json' in files
-            assert f'{root_name}/admin/cpu_profile_{n.account.hostname}-9644.json' in files
-
     @cluster(num_nodes=3)
     def test_get_config(self):
         node = self.redpanda.nodes[0]
diff --git a/tests/rptest/tests/rpk_debug_bundle_test.py b/tests/rptest/tests/rpk_debug_bundle_test.py
new file mode 100644
index 000000000000..f323211484a9
--- /dev/null
+++ b/tests/rptest/tests/rpk_debug_bundle_test.py
@@ -0,0 +1,192 @@
+# 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
+
+import os
+import re
+import zipfile
+import tempfile
+import random
+import string
+
+from ducktape.utils.util import wait_until
+from rptest.tests.redpanda_test import RedpandaTest
+from rptest.clients.rpk import RpkTool, RpkException
+from rptest.services.cluster import cluster
+from rptest.clients.rpk_remote import RpkRemoteTool
+
+
+class RpkDebugBundleTest(RedpandaTest):
+    def __init__(self, ctx):
+        super(RpkDebugBundleTest, self).__init__(test_context=ctx)
+        self._ctx = ctx
+        self._rpk = RpkTool(self.redpanda)
+
+    @cluster(num_nodes=3)
+    def test_debug_bundle(self):
+        # The main RpkTool helper runs rpk on the test runner machine -- debug
+        # commands are run on redpanda nodes.
+        root_name = "bundle" + ''.join(
+            random.choice(string.ascii_letters) for _ in range(5))
+        bundle_name = root_name + ".zip"
+        working_dir = "/tmp"
+        file_path = os.path.join(working_dir, bundle_name)
+        node = self.redpanda.nodes[0]
+
+        rpk_remote = RpkRemoteTool(self.redpanda, node)
+        output = rpk_remote.debug_bundle(file_path)
+        lines = output.split("\n")
+
+        # On error, rpk bundle returns 0 but writes error description to stdout
+        output_file = None
+        error_lines = []
+        any_errs = False
+        for l in lines:
+            self.logger.info(l)
+            if l.strip().startswith("* "):
+                error_lines.append(l)
+            elif 'errors occurred' in l:
+                any_errs = True
+            else:
+                m = re.match("^Debug bundle saved to '(.+)'$", l)
+                if m:
+                    output_file = m.group(1)
+
+        # Avoid false passes if our error line scraping gets broken
+        # by a format change.
+        if any_errs:
+            assert error_lines, f"Found error, but unable to parse error lines: {output}"
+
+        filtered_errors = []
+        for l in error_lines:
+            if "dmidecode" in l:
+                # dmidecode doesn't work in ducktape containers, ignore
+                # errors about it.
+                continue
+            if re.match(r".* error querying .*\.ntp\..* i\/o timeout", l):
+                self.logger.error(f"Non-fatal transitory NTP error: {l}")
+            else:
+                self.logger.error(f"Bad output line: {l}")
+                filtered_errors.append(l)
+
+        assert not filtered_errors, f"Unexpected errors encountered: {filtered_errors}"
+        assert output_file == file_path, f"Expected output file({output_file}) to be in {file_path}"
+
+        node.account.copy_from(output_file, working_dir)
+
+        zf = zipfile.ZipFile(output_file)
+        files = zf.namelist()
+        assert f'{root_name}/redpanda.yaml' in files, f"redpanda.yaml not found in the zip files: {files}"
+        assert f'{root_name}/redpanda.log' in files, f"redpanda.log not found in the zip files: {files}"
+
+        # At least the first controller log is being saved:
+        assert f'{root_name}/controller-logs/redpanda/controller/0_0/0-1-v1.log' in files, f"controller log (0-1-v1.log) not found in the zip files: {files}"
+
+        # Cluster admin API calls:
+        assert f'{root_name}/admin/brokers.json' in files, f"admin/brokers.json not found in the zip files: {files}"
+        assert f'{root_name}/admin/cluster_config.json' in files, f"admin/cluster_config.json not found in the zip files: {files}"
+        assert f'{root_name}/admin/health_overview.json' in files, f"admin/health_overview.json not found in the zip files: {files}"
+
+        # Per-node admin API calls:
+        for n in self.redpanda.started_nodes():
+            # rpk will save 2 snapsots per metrics endpoint:
+            assert f'{root_name}/metrics/{n.account.hostname}-9644/t0_metrics.txt' in files, f"/metrics/{n.account.hostname}-9644/t0_metrics.txt not found in the zip files: {files}"
+            assert f'{root_name}/metrics/{n.account.hostname}-9644/t1_metrics.txt' in files, f"/metrics/{n.account.hostname}-9644/t1_metrics.txt not found in the zip files: {files}"
+            assert f'{root_name}/metrics/{n.account.hostname}-9644/t0_public_metrics.txt' in files, f"/metrics/{n.account.hostname}-9644/t0_public_metrics.txt not found in the zip files: {files}"
+            assert f'{root_name}/metrics/{n.account.hostname}-9644/t1_public_metrics.txt' in files, f"/metrics/{n.account.hostname}-9644/t1_public_metrics.txt not found in the zip files: {files}"
+            # and 1 cluster_view and node_config per node:
+            assert f'{root_name}/admin/cluster_view_{n.account.hostname}-9644.json' in files, f"/admin/cluster_view_{n.account.hostname}-9644.json not found in the zip files: {files}"
+            assert f'{root_name}/admin/node_config_{n.account.hostname}-9644.json' in files, f"/admin/node_config_{n.account.hostname}-9644.json not found in the zip files: {files}"
+            assert f'{root_name}/admin/cpu_profile_{n.account.hostname}-9644.json' in files, f"/admin/cpu_profile_{n.account.hostname}-9644.json not found in the zip files: {files}"
+
+    @cluster(num_nodes=3)
+    def test_remote_debug_bundle_default(self):
+        """
+        e2e test, different scenarios using rpk default
+        values.
+        """
+        def _get_job_ID(output):
+            # By default, rpk creates a random UUID.
+            job_id_pattern = r'\b([0-9a-fA-F\-]{36})\b'
+
+            job_ids = re.findall(job_id_pattern, output)
+            assert len(job_ids) > 0, f"found no job-ID in: {output}"
+            assert len(
+                set(job_ids)) == 1, f"Not all job IDs are the same: {job_ids}"
+            return job_ids[0]
+
+        out = self._rpk.remote_bundle_start()
+        job_id = _get_job_ID(out)
+        assert "The debug bundle collection process has started" in out, f"unexpected output: {out}"
+
+        def _bundle_status(expected_status):
+            all_status = self._rpk.remote_bundle_status()
+            for s in all_status:
+                status = s["status"]
+                if status == 'error' and expected_status != 'error':
+                    raise RpkException(
+                        f"found error while creating a remote bundle: {s}")
+                if status != expected_status:
+                    return False
+            return True
+
+        wait_until(
+            lambda: _bundle_status("running"),
+            timeout_sec=60,
+            backoff_sec=1,
+            err_msg="Timed out waiting for debug bundle process to start")
+        # Cancel and test that all status are "error".
+        out = self._rpk.remote_bundle_cancel(job_id)
+        wait_until(
+            lambda: _bundle_status("error"),
+            timeout_sec=60,
+            backoff_sec=1,
+            err_msg="Timed out waiting for debug bundle process to be canceled"
+        )
+
+        # Start a new one
+        out = self._rpk.remote_bundle_start()
+        job_id = _get_job_ID(out)
+        assert "The debug bundle collection process has started" in out, f"unexpected output: {out}"
+
+        wait_until(
+            lambda: _bundle_status("success"),
+            timeout_sec=120,
+            backoff_sec=1,
+            err_msg="Timed out waiting for debug bundle process to finish")
+
+        def _validate_bundle_file(output_file):
+            # Open the outer zip file (the main bundle)
+            with zipfile.ZipFile(output_file) as zf:
+                # Get the list of files inside the outer zip file, one per broker.
+                files = zf.namelist()
+                assert len(
+                    files
+                ) == 3, f"Downloaded zip contains less files than the number of brokers: {files}"
+
+                for broker_file in files:
+                    with zf.open(broker_file) as inner_file:
+                        with zipfile.ZipFile(inner_file) as bzf:
+                            filename = os.path.split(bzf.filename)[1]
+                            root_name = os.path.splitext(filename)[0]
+                            bfiles = bzf.namelist()
+                            # We check if a sub-set of files are present:
+                            assert f'{root_name}/redpanda.yaml' in bfiles, f"{root_name}/redpanda.yaml not found in zip files: {bfiles}"
+                            # At least the first controller log is being saved:
+                            assert f'{root_name}/controller-logs/redpanda/controller/0_0/0-1-v1.log' in bfiles, f"{root_name}/controller-logs/redpanda/controller/0_0/0-1-v1.log not found in zip files: {bfiles}"
+                            # At least one cluster admin API call:
+                            assert f'{root_name}/admin/brokers.json' in bfiles, f"{root_name}/admin/brokers.json not found in zip files: {bfiles}"
+                            # Kafka metadata
+                            assert f'{root_name}/kafka.json' in bfiles, f"{root_name}/kafka.json not found in zip files: {bfiles}"
+
+        with tempfile.TemporaryDirectory() as td:
+            output_file = os.path.join(td, "test.zip")
+            out = self._rpk.remote_bundle_download(job_id,
+                                                   output_file=output_file)
+
+            _validate_bundle_file(output_file)