rpk: add 'rpk sql debug bundle' for Oxla diagnostics

src/go/rpk/pkg/cli/BUILD

@@ -31,6 +31,7 @@ go_library(
         "//src/go/rpk/pkg/cli/registry",
         "//src/go/rpk/pkg/cli/security",
         "//src/go/rpk/pkg/cli/shadow",
+        "//src/go/rpk/pkg/cli/sql",
         "//src/go/rpk/pkg/cli/topic",
         "//src/go/rpk/pkg/cli/transform",
         "//src/go/rpk/pkg/cli/version",

src/go/rpk/pkg/cli/root.go

@@ -37,6 +37,7 @@ import (
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/registry"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/security"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/shadow"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/sql"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/topic"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/transform"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/version"
@@ -152,6 +153,7 @@ Use --print-tree to emit the full command tree as JSON.`,
 		registry.NewCommand(fs, p),
 		security.NewCommand(fs, p),
 		shadow.NewCommand(fs, p),
+		sql.NewCommand(fs, p),
 		topic.NewCommand(fs, p),
 		transform.NewCommand(fs, p, osExec),
 		versioncmd.NewCommand(fs, p),

src/go/rpk/pkg/cli/sql/BUILD

@@ -0,0 +1,14 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "sql",
+    srcs = ["sql.go"],
+    importpath = "github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/sql",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//src/go/rpk/pkg/cli/sql/debug",
+        "//src/go/rpk/pkg/config",
+        "@com_github_spf13_afero//:afero",
+        "@com_github_spf13_cobra//:cobra",
+    ],
+)

src/go/rpk/pkg/cli/sql/debug/BUILD

@@ -0,0 +1,14 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "debug",
+    srcs = ["debug.go"],
+    importpath = "github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/sql/debug",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//src/go/rpk/pkg/cli/sql/debug/bundle",
+        "//src/go/rpk/pkg/config",
+        "@com_github_spf13_afero//:afero",
+        "@com_github_spf13_cobra//:cobra",
+    ],
+)

src/go/rpk/pkg/cli/sql/debug/bundle/BUILD

@@ -0,0 +1,20 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "bundle",
+    srcs = [
+        "bundle.go",
+        "client.go",
+        "collect.go",
+        "messages.go",
+    ],
+    importpath = "github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/sql/debug/bundle",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//src/go/rpk/pkg/cli/debug/debugbundle",
+        "//src/go/rpk/pkg/config",
+        "//src/go/rpk/pkg/out",
+        "@com_github_spf13_afero//:afero",
+        "@com_github_spf13_cobra//:cobra",
+    ],
+)

src/go/rpk/pkg/cli/sql/debug/bundle/bundle.go

@@ -0,0 +1,185 @@
+// Copyright 2026 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 bundle
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/debug/debugbundle"
+	"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"
+)
+
+const bundleHelpText = `Collect a read-only diagnostic bundle from an Oxla (SQL) cluster.
+
+The bundle is gathered over the Oxla admin API (oxla.admin.v1.DebugService). It
+seeds from the first --admin-hosts entry, discovers the rest of the cluster via
+GetClusterNodes, and fans out per-node collection (config, logs, host probes,
+resource usage, a Prometheus /metrics time series) plus cluster-wide artifacts
+(topology, catalog head). Results are written as a ZIP under an sql/ subtree with
+a manifest.json and a per-RPC errors.txt roll-up; one failing RPC or unreachable
+node does not fail the bundle.
+
+Run it from inside an Oxla pod (it defaults to localhost:9090 and discovers the
+rest), or against a remote cluster by passing seed --admin-hosts.`
+
+func NewCommand(fs afero.Fs, _ *config.Params) *cobra.Command {
+	var (
+		adminHosts  []string
+		output      string
+		uploadURL   string
+		useTLS      bool
+		tlsSkip     bool
+		tlsCA       string
+		tlsCert     string
+		tlsKey      string
+		user        string
+		password    string
+		token       string
+		sqlText     string
+		vmstat      bool
+		cpuSeconds  uint
+		logSince    time.Duration
+		logSizeLim  uint64
+		metricsPort uint16
+		timeout     time.Duration
+	)
+	cmd := &cobra.Command{
+		Use:   "bundle",
+		Short: "Collect a diagnostic bundle from an Oxla (SQL) cluster",
+		Long:  bundleHelpText,
+		Args:  cobra.NoArgs,
+		Run: func(cmd *cobra.Command, _ []string) {
+			if len(adminHosts) == 0 {
+				adminHosts = []string{"localhost:9090"}
+			}
+
+			sqlMode, err := sqlTextMode(sqlText)
+			out.MaybeDieErr(err)
+
+			var tlsCfg *tls.Config
+			if useTLS {
+				tlsCfg, err = buildTLSConfig(fs, tlsSkip, tlsCA, tlsCert, tlsKey)
+				out.MaybeDie(err, "unable to build TLS config: %v", err)
+			}
+
+			var logSinceMs int64
+			if logSince > 0 {
+				logSinceMs = time.Now().Add(-logSince).UnixMilli()
+			}
+
+			// The local artifact is always written first; upload is a separate
+			// step, so a failed upload still leaves the bundle on disk.
+			f, err := fs.Create(output)
+			out.MaybeDie(err, "unable to create %q: %v", output, err)
+
+			opts := Options{
+				Seeds:             adminHosts,
+				UseTLS:            useTLS,
+				TLSConfig:         tlsCfg,
+				Auth:              Auth{Bearer: token, User: user, Pass: password},
+				Timeout:           timeout,
+				SQLTextMode:       sqlMode,
+				IncludeVmstat:     vmstat,
+				CPUProfileSeconds: uint32(cpuSeconds),
+				LogSinceUnixMs:    logSinceMs,
+				LogSizeLimitBytes: logSizeLim,
+				MetricsPort:       metricsPort,
+				ToolVersion:       "rpk",
+			}
+			runErr := writeBundle(cmd.Context(), f, opts)
+			closeErr := f.Close()
+			out.MaybeDie(runErr, "unable to collect debug bundle: %v", runErr)
+			out.MaybeDie(closeErr, "unable to finalize %q: %v", output, closeErr)
+			fmt.Printf("Wrote debug bundle to %q\n", output)
+
+			if uploadURL != "" {
+				err = debugbundle.UploadBundle(cmd.Context(), output, uploadURL)
+				out.MaybeDie(err, "unable to upload bundle: %v", err)
+				fmt.Println("Successfully uploaded the bundle")
+			}
+		},
+	}
+
+	f := cmd.Flags()
+	f.StringSliceVar(&adminHosts, "admin-hosts", nil, "Comma-separated seed admin endpoints host:port; the rest of the cluster is discovered via GetClusterNodes (default localhost:9090)")
+	f.StringVarP(&output, "output", "o", "oxla-debug-bundle.zip", "Output ZIP path")
+	f.StringVar(&uploadURL, "upload-url", "", "If provided, where to upload the bundle in addition to creating a copy on disk")
+	f.BoolVar(&useTLS, "tls", false, "Use HTTPS for admin endpoints")
+	f.BoolVar(&tlsSkip, "tls-insecure-skip-verify", false, "Skip TLS certificate verification")
+	f.StringVar(&tlsCA, "tls-ca", "", "PEM CA bundle for server verification")
+	f.StringVar(&tlsCert, "tls-cert", "", "Client certificate for mTLS")
+	f.StringVar(&tlsKey, "tls-key", "", "Client key for mTLS")
+	f.StringVar(&user, "user", "", "HTTP Basic username")
+	f.StringVar(&password, "password", "", "HTTP Basic password")
+	f.StringVar(&token, "token", "", "Bearer token (wins over Basic)")
+	f.StringVar(&sqlText, "include-sql-text", "masked", "SQL text in query artifacts: masked|raw")
+	f.BoolVar(&vmstat, "include-vmstat", false, "Include vmstat in host probes (~1s slower)")
+	f.UintVar(&cpuSeconds, "cpu-profile-seconds", 0, "Collect a CPU profile of this duration per node (0 = skip)")
+	f.DurationVar(&logSince, "log-since", 0, "Collect log lines newer than this (0 = server default window)")
+	f.Uint64Var(&logSizeLim, "log-size-limit", 0, "Max log bytes per node (0 = server default)")
+	f.Uint16Var(&metricsPort, "metrics-port", 8080, "Per-node Prometheus metrics port (scraped twice ~1s apart)")
+	f.DurationVar(&timeout, "timeout", 60*time.Second, "Per-RPC timeout")
+	return cmd
+}
+
+func sqlTextMode(s string) (string, error) {
+	switch strings.ToLower(s) {
+	case "masked":
+		return "SQL_TEXT_MODE_MASKED", nil
+	case "raw":
+		return "SQL_TEXT_MODE_RAW", nil
+	default:
+		return "", fmt.Errorf("--include-sql-text must be masked|raw, got %q", s)
+	}
+}
+
+func buildTLSConfig(fs afero.Fs, skipVerify bool, caFile, certFile, keyFile string) (*tls.Config, error) {
+	cfg := &tls.Config{
+		MinVersion:         tls.VersionTLS12,
+		InsecureSkipVerify: skipVerify,
+	}
+	if caFile != "" {
+		pem, err := afero.ReadFile(fs, caFile)
+		if err != nil {
+			return nil, fmt.Errorf("unable to read --tls-ca: %w", err)
+		}
+		pool := x509.NewCertPool()
+		if !pool.AppendCertsFromPEM(pem) {
+			return nil, fmt.Errorf("--tls-ca: no certificates found in %s", caFile)
+		}
+		cfg.RootCAs = pool
+	}
+	if (certFile == "") != (keyFile == "") {
+		return nil, fmt.Errorf("--tls-cert and --tls-key must be set together")
+	}
+	if certFile != "" {
+		certPEM, err := afero.ReadFile(fs, certFile)
+		if err != nil {
+			return nil, fmt.Errorf("unable to read --tls-cert: %w", err)
+		}
+		keyPEM, err := afero.ReadFile(fs, keyFile)
+		if err != nil {
+			return nil, fmt.Errorf("unable to read --tls-key: %w", err)
+		}
+		cert, err := tls.X509KeyPair(certPEM, keyPEM)
+		if err != nil {
+			return nil, fmt.Errorf("unable to load client cert/key: %w", err)
+		}
+		cfg.Certificates = []tls.Certificate{cert}
+	}
+	return cfg, nil
+}

src/go/rpk/pkg/cli/sql/debug/bundle/client.go

@@ -0,0 +1,135 @@
+// Copyright 2026 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 bundle
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+)
+
+// servicePath is the Connect route prefix for oxla.admin.v1.DebugService. Method
+// names are appended verbatim, e.g. .../GetVersion.
+const servicePath = "/oxla.admin.v1.DebugService/"
+
+// maxResponseBytes caps a single RPC response to guard against a hostile or
+// wedged server. Debug payloads (log tails, CPU profiles) are large but bounded.
+const maxResponseBytes = 1 << 30 // 1 GiB
+
+// Auth carries the credentials sent on every request. Bearer wins over Basic
+// when both are set.
+type Auth struct {
+	Bearer string
+	User   string
+	Pass   string
+}
+
+func (a Auth) apply(req *http.Request) {
+	switch {
+	case a.Bearer != "":
+		req.Header.Set("Authorization", "Bearer "+a.Bearer)
+	case a.User != "" || a.Pass != "":
+		req.SetBasicAuth(a.User, a.Pass)
+	}
+}
+
+// ConnectError is the Connect unary error envelope returned on non-2xx: a JSON
+// body of {"code","message"}. httpStatus is retained for cases where the body is
+// absent or not a Connect envelope (e.g. a 404 from the router).
+type ConnectError struct {
+	Code       string `json:"code"`
+	Message    string `json:"message"`
+	httpStatus int
+}
+
+func (e *ConnectError) Error() string {
+	if e.Code != "" {
+		return fmt.Sprintf("connect error (%s): %s", e.Code, e.Message)
+	}
+	if e.Message != "" {
+		return fmt.Sprintf("http %d: %s", e.httpStatus, e.Message)
+	}
+	return fmt.Sprintf("http %d", e.httpStatus)
+}
+
+// Client is a minimal Connect unary client for a single admin endpoint. It speaks
+// the application/json codec so the tool has no protobuf dependency.
+type Client struct {
+	baseURL string
+	hc      *http.Client
+	auth    Auth
+}
+
+func NewClient(baseURL string, hc *http.Client, auth Auth) *Client {
+	return &Client{baseURL: baseURL, hc: hc, auth: auth}
+}
+
+// CallRaw invokes one unary method and returns the response body as raw JSON.
+// req is marshaled as the request message; pass emptyRequest for no-field
+// requests.
+func (c *Client) CallRaw(ctx context.Context, method string, req any) (json.RawMessage, error) {
+	reqBody, err := json.Marshal(req)
+	if err != nil {
+		return nil, fmt.Errorf("marshal %s request: %w", method, err)
+	}
+
+	url := c.baseURL + servicePath + method
+	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(reqBody))
+	if err != nil {
+		return nil, fmt.Errorf("build %s request: %w", method, err)
+	}
+	httpReq.Header.Set("Content-Type", "application/json")
+	httpReq.Header.Set("Accept", "application/json")
+	httpReq.Header.Set("Connect-Protocol-Version", "1")
+	c.auth.apply(httpReq)
+
+	resp, err := c.hc.Do(httpReq)
+	if err != nil {
+		// The transport error already names the method via the request URL;
+		// the caller supplies node/RPC context when recording it.
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes))
+	if err != nil {
+		return nil, fmt.Errorf("%s: read response: %w", method, err)
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		ce := &ConnectError{httpStatus: resp.StatusCode}
+		if jsonErr := json.Unmarshal(body, ce); jsonErr != nil || (ce.Code == "" && ce.Message == "") {
+			ce.Message = string(bytes.TrimSpace(body))
+		}
+		return nil, ce
+	}
+	return json.RawMessage(body), nil
+}
+
+// Call invokes a method and unmarshals the response into out.
+func (c *Client) Call(ctx context.Context, method string, req, out any) error {
+	raw, err := c.CallRaw(ctx, method, req)
+	if err != nil {
+		return err
+	}
+	if out == nil {
+		return nil
+	}
+	if err := json.Unmarshal(raw, out); err != nil {
+		return fmt.Errorf("%s: decode response: %w", method, err)
+	}
+	return nil
+}
+
+// emptyRequest marshals to `{}`, the body for no-field request messages.
+var emptyRequest = struct{}{}