From f606a2390ebd27783a11c03bb0d5479dfbf3fe83 Mon Sep 17 00:00:00 2001 From: Wojciech Oziebly Date: Mon, 13 Jul 2026 16:01:37 +0200 Subject: [PATCH] rpk: add 'rpk sql debug bundle' for Redpanda SQL diagnostics --- src/go/rpk/pkg/cli/BUILD | 1 + src/go/rpk/pkg/cli/root.go | 2 + src/go/rpk/pkg/cli/sql/BUILD | 14 + src/go/rpk/pkg/cli/sql/debug/BUILD | 14 + src/go/rpk/pkg/cli/sql/debug/bundle/BUILD | 22 + src/go/rpk/pkg/cli/sql/debug/bundle/bundle.go | 210 ++++++++ src/go/rpk/pkg/cli/sql/debug/bundle/client.go | 114 +++++ .../rpk/pkg/cli/sql/debug/bundle/collect.go | 447 ++++++++++++++++++ .../rpk/pkg/cli/sql/debug/bundle/messages.go | 133 ++++++ src/go/rpk/pkg/cli/sql/debug/debug.go | 29 ++ src/go/rpk/pkg/cli/sql/sql.go | 30 ++ 11 files changed, 1016 insertions(+) create mode 100644 src/go/rpk/pkg/cli/sql/BUILD create mode 100644 src/go/rpk/pkg/cli/sql/debug/BUILD create mode 100644 src/go/rpk/pkg/cli/sql/debug/bundle/BUILD create mode 100644 src/go/rpk/pkg/cli/sql/debug/bundle/bundle.go create mode 100644 src/go/rpk/pkg/cli/sql/debug/bundle/client.go create mode 100644 src/go/rpk/pkg/cli/sql/debug/bundle/collect.go create mode 100644 src/go/rpk/pkg/cli/sql/debug/bundle/messages.go create mode 100644 src/go/rpk/pkg/cli/sql/debug/debug.go create mode 100644 src/go/rpk/pkg/cli/sql/sql.go diff --git a/src/go/rpk/pkg/cli/BUILD b/src/go/rpk/pkg/cli/BUILD index 96737b462105b..28c3369acc49d 100644 --- a/src/go/rpk/pkg/cli/BUILD +++ b/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", diff --git a/src/go/rpk/pkg/cli/root.go b/src/go/rpk/pkg/cli/root.go index a34c476718caf..6b75563a9bfd3 100644 --- a/src/go/rpk/pkg/cli/root.go +++ b/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), diff --git a/src/go/rpk/pkg/cli/sql/BUILD b/src/go/rpk/pkg/cli/sql/BUILD new file mode 100644 index 0000000000000..348b4b46bb0d7 --- /dev/null +++ b/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", + ], +) diff --git a/src/go/rpk/pkg/cli/sql/debug/BUILD b/src/go/rpk/pkg/cli/sql/debug/BUILD new file mode 100644 index 0000000000000..0f2baff1ca788 --- /dev/null +++ b/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", + ], +) diff --git a/src/go/rpk/pkg/cli/sql/debug/bundle/BUILD b/src/go/rpk/pkg/cli/sql/debug/bundle/BUILD new file mode 100644 index 0000000000000..c5cf9d2cf6539 --- /dev/null +++ b/src/go/rpk/pkg/cli/sql/debug/bundle/BUILD @@ -0,0 +1,22 @@ +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/httpapi", + "//src/go/rpk/pkg/out", + "@com_github_spf13_afero//:afero", + "@com_github_spf13_cobra//:cobra", + "@com_github_spf13_pflag//:pflag", + ], +) diff --git a/src/go/rpk/pkg/cli/sql/debug/bundle/bundle.go b/src/go/rpk/pkg/cli/sql/debug/bundle/bundle.go new file mode 100644 index 0000000000000..edab0bdf6262a --- /dev/null +++ b/src/go/rpk/pkg/cli/sql/debug/bundle/bundle.go @@ -0,0 +1,210 @@ +// 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" + "github.com/spf13/pflag" +) + +const bundleHelpText = `Collect a read-only diagnostic bundle from a Redpanda SQL cluster. + +The bundle is gathered over the Redpanda SQL admin API. It seeds from the first +--admin-hosts entry (defaulting to localhost:9090), +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 a Redpanda SQL pod (it defaults to localhost:9090 and discovers +the rest), or against a remote cluster by passing seed --admin-hosts.` + +type tlsFlags struct { + enabled bool + skipVerify bool + ca string + cert string + key string +} + +type authFlags struct { + user string + password string + token string +} + +type bundleFlags struct { + adminHosts []string + output string + uploadURL string + tls tlsFlags + auth authFlags + sqlText string + vmstat bool + cpuSeconds uint + logSince time.Duration + logSizeLim uint64 + metricsPort uint16 + timeout time.Duration +} + +func NewCommand(fs afero.Fs, _ *config.Params) *cobra.Command { + var cfg bundleFlags + cmd := &cobra.Command{ + Use: "bundle", + Short: "Collect a diagnostic bundle from a Redpanda SQL cluster", + Long: bundleHelpText, + Args: cobra.NoArgs, + Run: func(cmd *cobra.Command, _ []string) { + opts, err := cfg.options(fs) + out.MaybeDieErr(err) + + // 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(cfg.output) + out.MaybeDie(err, "unable to create %q: %v", cfg.output, err) + + 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", cfg.output, closeErr) + fmt.Printf("Wrote debug bundle to %q\n", cfg.output) + + if cfg.uploadURL != "" { + err = debugbundle.UploadBundle(cmd.Context(), cfg.output, cfg.uploadURL) + out.MaybeDie(err, "unable to upload bundle: %v", err) + fmt.Println("Successfully uploaded the bundle") + } + }, + } + cfg.install(cmd.Flags()) + + cmd.MarkFlagsRequiredTogether("tls-cert", "tls-key") + cmd.MarkFlagsRequiredTogether("user", "password") + cmd.MarkFlagsMutuallyExclusive("user", "token") + cmd.MarkFlagsMutuallyExclusive("password", "token") + return cmd +} + +// install registers every bundle flag against the flag set. +func (c *bundleFlags) install(f *pflag.FlagSet) { + f.StringSliceVar(&c.adminHosts, "admin-hosts", []string{"localhost:9090"}, "Comma-separated seed admin endpoints host:port; the rest of the cluster is discovered via GetClusterNodes") + f.StringVarP(&c.output, "output", "o", "redpanda-sql-debug-bundle.zip", "Output ZIP path") + f.StringVar(&c.uploadURL, "upload-url", "", "If provided, where to upload the bundle in addition to creating a copy on disk") + f.BoolVar(&c.tls.enabled, "tls", false, "Use HTTPS for admin endpoints") + f.BoolVar(&c.tls.skipVerify, "tls-insecure-skip-verify", false, "Skip TLS certificate verification") + f.StringVar(&c.tls.ca, "tls-ca", "", "PEM CA bundle for server verification") + f.StringVar(&c.tls.cert, "tls-cert", "", "Client certificate for mTLS") + f.StringVar(&c.tls.key, "tls-key", "", "Client key for mTLS") + f.StringVar(&c.auth.user, "user", "", "HTTP Basic username") + f.StringVar(&c.auth.password, "password", "", "HTTP Basic password") + f.StringVar(&c.auth.token, "token", "", "Bearer token (mutually exclusive with --user/--password)") + f.StringVar(&c.sqlText, "include-sql-text", "masked", "SQL text in query artifacts: masked|raw") + f.BoolVar(&c.vmstat, "include-vmstat", false, "Include vmstat in host probes (~1s slower)") + f.UintVar(&c.cpuSeconds, "cpu-profile-seconds", 0, "Collect a CPU profile of this duration per node (0 = skip)") + f.DurationVar(&c.logSince, "log-since", 0, "Collect log lines newer than this (0 = server default window)") + f.Uint64Var(&c.logSizeLim, "log-size-limit", 0, "Max log bytes per node (0 = server default)") + f.Uint16Var(&c.metricsPort, "metrics-port", 8080, "Per-node Prometheus metrics port (scraped twice ~1s apart)") + f.DurationVar(&c.timeout, "timeout", 60*time.Second, "Per-RPC timeout") +} + +// options validates the flags and assembles the collector Options. +func (c *bundleFlags) options(fs afero.Fs) (Options, error) { + sqlMode, err := sqlTextMode(c.sqlText) + if err != nil { + return Options{}, err + } + + var tlsCfg *tls.Config + if c.tls.enabled { + tlsCfg, err = c.tls.config(fs) + if err != nil { + return Options{}, fmt.Errorf("unable to build TLS config: %w", err) + } + } + + var logSinceMs int64 + if c.logSince > 0 { + logSinceMs = time.Now().Add(-c.logSince).UnixMilli() + } + + return Options{ + Seeds: c.adminHosts, + UseTLS: c.tls.enabled, + TLSConfig: tlsCfg, + Auth: Auth{Bearer: c.auth.token, User: c.auth.user, Pass: c.auth.password}, + Timeout: c.timeout, + SQLTextMode: sqlMode, + IncludeVmstat: c.vmstat, + CPUProfileSeconds: uint32(c.cpuSeconds), + LogSinceUnixMs: logSinceMs, + LogSizeLimitBytes: c.logSizeLim, + MetricsPort: c.metricsPort, + ToolVersion: "rpk", + }, nil +} + +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) + } +} + +// config builds a *tls.Config from the flags. --tls-cert and --tls-key are +// enforced together by the command's MarkFlagsRequiredTogether. +func (t tlsFlags) config(fs afero.Fs) (*tls.Config, error) { + cfg := &tls.Config{ + MinVersion: tls.VersionTLS12, + InsecureSkipVerify: t.skipVerify, + } + if t.ca != "" { + pem, err := afero.ReadFile(fs, t.ca) + 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", t.ca) + } + cfg.RootCAs = pool + } + if t.cert != "" { + certPEM, err := afero.ReadFile(fs, t.cert) + if err != nil { + return nil, fmt.Errorf("unable to read --tls-cert: %w", err) + } + keyPEM, err := afero.ReadFile(fs, t.key) + 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 +} diff --git a/src/go/rpk/pkg/cli/sql/debug/bundle/client.go b/src/go/rpk/pkg/cli/sql/debug/bundle/client.go new file mode 100644 index 0000000000000..82974fa2c8aad --- /dev/null +++ b/src/go/rpk/pkg/cli/sql/debug/bundle/client.go @@ -0,0 +1,114 @@ +// 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 ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + + "github.com/redpanda-data/redpanda/src/go/rpk/pkg/httpapi" +) + +const servicePath = "/oxla.admin.v1.DebugService/" + +// 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 +} + +// 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) +} + +type Client struct { + cl *httpapi.Client +} + +func NewClient(baseURL string, hc *http.Client, auth Auth) *Client { + opts := []httpapi.Opt{ + httpapi.Host(baseURL), + httpapi.HTTPClient(hc), + // One attempt per RPC: a failing or unreachable node is recorded and + // the bundle moves on, rather than stalling on retry backoff. + httpapi.Retries(0), + httpapi.Headers("Connect-Protocol-Version", "1"), + // Connect returns its {"code","message"} envelope alongside a mapped + // HTTP status; decode 4xx bodies into it for a useful error string. + httpapi.Err4xx(func(status int) error { return &ConnectError{httpStatus: status} }), + } + switch { + case auth.Bearer != "": + opts = append(opts, httpapi.BearerAuth(auth.Bearer)) + case auth.User != "" || auth.Pass != "": + opts = append(opts, httpapi.BasicAuth(auth.User, auth.Pass)) + } + return &Client{cl: httpapi.NewClient(opts...)} +} + +// 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) { + var raw []byte + if err := c.cl.Post(ctx, servicePath+method, nil, "application/json", req, &raw); err != nil { + return nil, connectErr(err) + } + return json.RawMessage(raw), 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 +} + +// connectErr normalizes a non-4xx httpapi status error (e.g. a 5xx Connect +// error) into a ConnectError so recorded messages stay uniform; 4xx bodies +// already arrive as a ConnectError via the Err4xx hook. +func connectErr(err error) error { + var be *httpapi.BodyError + if errors.As(err, &be) { + return &ConnectError{httpStatus: be.StatusCode} + } + return err +} + +// emptyRequest marshals to `{}`, the body for no-field request messages. +var emptyRequest = struct{}{} diff --git a/src/go/rpk/pkg/cli/sql/debug/bundle/collect.go b/src/go/rpk/pkg/cli/sql/debug/bundle/collect.go new file mode 100644 index 0000000000000..cca011841c84e --- /dev/null +++ b/src/go/rpk/pkg/cli/sql/debug/bundle/collect.go @@ -0,0 +1,447 @@ +// 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 ( + "archive/zip" + "context" + "crypto/tls" + "encoding/json" + "fmt" + "io" + "net/http" + "strconv" + "strings" + "time" + + "github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/debug/debugbundle" +) + +// bundleRoot is the top-level subtree for all Redpanda SQL artifacts, matching the +// design's `sql/` layout so the output can be spliced under an rpk bundle. +const bundleRoot = "sql" + +// maxResponseBytes caps the metrics scrape response to guard against a hostile +// or wedged server. Prometheus payloads are large but bounded. +const maxResponseBytes = 1 << 30 // 1 GiB + +// Options configures a collection run. +type Options struct { + Seeds []string + UseTLS bool + TLSConfig *tls.Config + Auth Auth + Timeout time.Duration + SQLTextMode string // sqlTextModeMasked | sqlTextModeRaw + IncludeVmstat bool + CPUProfileSeconds uint32 + LogSinceUnixMs int64 + LogSizeLimitBytes uint64 + MetricsPort uint16 + ToolVersion string +} + +// collectionResult is one per-RPC, per-node outcome recorded in the manifest. +type collectionResult struct { + Node string `json:"node"` + RPC string `json:"rpc"` + Status string `json:"status"` // "ok" | "error" + ElapsedMs int64 `json:"elapsed_ms"` + Error string `json:"error,omitempty"` +} + +type manifest struct { + BundleCreatedAt string `json:"bundle_created_at"` + ToolVersion string `json:"tool_version"` + RedpandaSQLVersions map[string]string `json:"redpanda_sql_versions"` + NodesAttempted int `json:"nodes_attempted"` + NodesSucceeded int `json:"nodes_succeeded"` + CollectionResults []collectionResult `json:"collection_results"` + RedactionModes map[string]string `json:"redaction_modes"` +} + +// Run collects a debug bundle and streams it as a ZIP into out. +func writeBundle(ctx context.Context, out io.Writer, opts Options) error { + b := &bundle{ + zw: zip.NewWriter(out), + opts: opts, + hc: &http.Client{ + Timeout: opts.Timeout, + Transport: &http.Transport{TLSClientConfig: opts.TLSConfig}, + }, + versions: map[string]string{}, + } + b.collect(ctx) + b.writeManifest() + b.writeErrors() + return b.zw.Close() +} + +type bundle struct { + zw *zip.Writer + opts Options + hc *http.Client + results []collectionResult + errs []string + versions map[string]string // endpoint -> version +} + +func (b *bundle) scheme() string { + if b.opts.UseTLS { + return "https" + } + return "http" +} + +func (b *bundle) client(endpoint string) *Client { + return NewClient(b.scheme()+"://"+endpoint, b.hc, b.opts.Auth) +} + +func (b *bundle) collect(ctx context.Context) { + endpoints, clusterCl := b.discover(ctx) + b.clusterCalls(ctx, clusterCl) + for _, ep := range endpoints { + b.nodeCalls(ctx, ep) + } +} + +// discover resolves the node list from the first seed that answers +// GetClusterNodes, writing cluster_nodes.json as a side effect. It falls back to +// treating the seeds themselves as the node list so a single wedged node (or one +// that lacks GetClusterNodes) still yields a bundle. Returns the endpoints to fan +// out to and the client to use for cluster-wide calls. +func (b *bundle) discover(ctx context.Context) ([]string, *Client) { + for _, seed := range b.opts.Seeds { + cl := b.client(seed) + start := time.Now() + raw, err := cl.CallRaw(ctx, "GetClusterNodes", emptyRequest) + b.record(seed, "GetClusterNodes", start, err) + if err != nil { + continue + } + b.addJSON(bundleRoot+"/cluster/cluster_nodes.json", raw) + + var resp getClusterNodesResponse + if json.Unmarshal(raw, &resp) == nil && len(resp.Nodes) > 0 { + endpoints := make([]string, 0, len(resp.Nodes)) + for _, n := range resp.Nodes { + if n.AdminEndpoint != "" { + endpoints = append(endpoints, n.AdminEndpoint) + } + } + if len(endpoints) > 0 { + return endpoints, cl + } + } + return b.opts.Seeds, cl + } + // No seed answered discovery; fan out to the seeds directly. + return b.opts.Seeds, b.client(b.opts.Seeds[0]) +} + +// clusterCalls collects the cluster-wide artifacts once, from a single node. +func (b *bundle) clusterCalls(ctx context.Context, cl *Client) { + const node = "cluster" + dir := bundleRoot + "/cluster/" + + b.grabJSON(ctx, cl, node, "GetOxlaHomeListing", emptyRequest, dir+"redpanda_sql_home_listing.json") + b.grabJSON(ctx, cl, node, "GetRecentQueries", + getRecentQueriesRequest{IncludeSQLText: b.opts.SQLTextMode}, dir+"recent_queries.json") + + if head := (getCatalogHeadResponse{}); b.call(ctx, cl, node, "GetCatalogHead", emptyRequest, &head) { + b.add(dir+"catalog_head.pb", head.CatalogHead) + if head.CatalogHeadJSON != "" { + b.addJSON(dir+"catalog_head.json", json.RawMessage(head.CatalogHeadJSON)) + } + } +} + +// nodeCalls collects the per-node artifacts for one admin endpoint. +func (b *bundle) nodeCalls(ctx context.Context, endpoint string) { + cl := b.client(endpoint) + dir := fmt.Sprintf("%s/nodes/%s/", bundleRoot, debugbundle.SanitizeName(endpoint)) + + if raw, ok := b.grabJSON(ctx, cl, endpoint, "GetVersion", emptyRequest, dir+"version.json"); ok { + var v getVersionResponse + if json.Unmarshal(raw, &v) == nil { + b.versions[endpoint] = v.Version + } + } + // config.yaml is per-node (env overrides, host_name, ports differ), so collect + // it on every node rather than once cluster-wide. + if cfg := (getConfigResponse{}); b.call(ctx, cl, endpoint, "GetConfig", emptyRequest, &cfg) { + b.add(dir+"config.yaml", []byte(cfg.YAML)) + } + b.grabJSON(ctx, cl, endpoint, "GetActiveQueries", + getActiveQueriesRequest{IncludeSQLText: b.opts.SQLTextMode}, dir+"active_queries.json") + + b.resourceUsage(ctx, cl, endpoint, dir) + + if lt := (getLogTailResponse{}); b.call(ctx, cl, endpoint, "GetLogTail", + getLogTailRequest{SinceUnixMs: b.opts.LogSinceUnixMs, SizeLimitBytes: b.opts.LogSizeLimitBytes}, <) { + b.add(dir+"tail.log", lt.Content) + } + b.startupLog(ctx, cl, endpoint, dir) + b.crashReports(ctx, cl, endpoint, dir) + b.hostProbes(ctx, cl, endpoint, dir) + + if b.opts.CPUProfileSeconds > 0 { + if cp := (getCPUProfileResponse{}); b.call(ctx, cl, endpoint, "GetCpuProfile", + getCPUProfileRequest{DurationSeconds: b.opts.CPUProfileSeconds}, &cp) { + b.add(dir+"cpu_profile.pprof.gz", cp.PprofGzip) + } + } + b.scrapeMetrics(ctx, endpoint, dir) +} + +// scrapeMetrics pulls the node's Prometheus /metrics endpoint twice ~1s apart so +// the bundle carries a short time series for rate/delta analysis. Metrics are +// served by a separate plain-HTTP server (config `metrics.port`), not the admin +// API, so this uses http:// on that port and neither TLS nor auth. +func (b *bundle) scrapeMetrics(ctx context.Context, node, dir string) { + host := node + if i := strings.LastIndex(node, ":"); i >= 0 { + host = node[:i] + } + url := fmt.Sprintf("http://%s:%d/metrics", host, b.opts.MetricsPort) + + scrape := func(suffix string) { + start := time.Now() + err := b.httpGetToFile(ctx, url, dir+"metrics_"+suffix+".txt") + b.record(node, "GET /metrics", start, err) + } + scrape("t0") + time.Sleep(time.Second) + scrape("t1") +} + +// httpGetToFile GETs url and writes the body to name in the bundle. Non-200 is an +// error and nothing is written. +func (b *bundle) httpGetToFile(ctx context.Context, url, name string) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return err + } + resp, err := b.hc.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("GET %s: HTTP %d", url, resp.StatusCode) + } + body, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes)) + if err != nil { + return fmt.Errorf("GET %s: read body: %w", url, err) + } + b.add(name, body) + return nil +} + +func (b *bundle) resourceUsage(ctx context.Context, cl *Client, node, dir string) { + sample := func() (json.RawMessage, bool) { + start := time.Now() + raw, err := cl.CallRaw(ctx, "GetResourceUsage", emptyRequest) + b.record(node, "GetResourceUsage", start, err) + return raw, err == nil + } + first, ok := sample() + if !ok { + return + } + time.Sleep(time.Second) + second, ok := sample() + if !ok { + b.addJSON(dir+"resource-usage.json", first) + return + } + + report := map[string]any{"samples": []json.RawMessage{first, second}} + if pct, ok := cpuPercentage(first, second); ok { + report["cpuPercentage"] = pct + } + out, _ := json.MarshalIndent(report, "", " ") + b.add(dir+"resource-usage.json", out) +} + +func (b *bundle) startupLog(ctx context.Context, cl *Client, node, dir string) { + var resp getStartupLogResponse + if !b.call(ctx, cl, node, "GetStartupLog", emptyRequest, &resp) { + return + } + var sb strings.Builder + for _, f := range resp.Files { // newest-first + fmt.Fprintf(&sb, "===== %s =====\n", f.Filename) + sb.Write(f.Content) + sb.WriteByte('\n') + } + b.add(dir+"startup.log", []byte(sb.String())) +} + +func (b *bundle) crashReports(ctx context.Context, cl *Client, node, dir string) { + var resp getCrashReportsResponse + if !b.call(ctx, cl, node, "GetCrashReports", emptyRequest, &resp) { + return + } + for i, r := range resp.Reports { + name := r.Filename + if name == "" { + name = fmt.Sprintf("%s_%s.txt", nonempty(r.TimestampUnixMs, strconv.Itoa(i)), r.PID) + } + b.add(dir+"crash_reports/"+debugbundle.SanitizeName(name), []byte(r.Trace)) + } +} + +func (b *bundle) hostProbes(ctx context.Context, cl *Client, node, dir string) { + var resp getHostProbesResponse + if !b.call(ctx, cl, node, "GetHostProbes", getHostProbesRequest{IncludeVmstat: b.opts.IncludeVmstat}, &resp) { + return + } + files := map[string]string{ + "proc/cpuinfo": resp.ProcCPUInfo, + "proc/meminfo": resp.ProcMemInfo, + "proc/diskstats": resp.ProcDiskstats, + "proc/loadavg": resp.ProcLoadavg, + "proc/version": resp.ProcVersion, + "proc/mounts": resp.ProcMounts, + "proc/net/sockstat": resp.ProcNetSockstat, + "linux-utils/uname.txt": resp.Uname, + "linux-utils/df.txt": resp.DF, + "linux-utils/free.txt": resp.Free, + "linux-utils/vmstat.txt": resp.Vmstat, + "linux-utils/top.txt": resp.Top, + "linux-utils/uptime.txt": resp.Uptime, + "linux-utils/sysctl.txt": resp.Sysctl, + "linux-utils/resolv.txt": resp.ResolvConf, + } + for rel, content := range files { + if content != "" { + b.add(dir+rel, []byte(content)) + } + } + if resp.ContainerImageTag != "" || resp.ContainerImageDigst != "" { + b.add(dir+"linux-utils/container.txt", + []byte(fmt.Sprintf("tag: %s\ndigest: %s\n", resp.ContainerImageTag, resp.ContainerImageDigst))) + } +} + +// --- helpers ---------------------------------------------------------------- + +// grabJSON calls a method and writes its response verbatim (pretty-printed) to +// filename. Returns the raw response and whether the call succeeded. +func (b *bundle) grabJSON(ctx context.Context, cl *Client, node, method string, req any, filename string) (json.RawMessage, bool) { + start := time.Now() + raw, err := cl.CallRaw(ctx, method, req) + b.record(node, method, start, err) + if err != nil { + return nil, false + } + b.addJSON(filename, raw) + return raw, true +} + +// call invokes a method, records the outcome, and decodes into out. Returns +// whether the call succeeded. +func (b *bundle) call(ctx context.Context, cl *Client, node, method string, req, out any) bool { + start := time.Now() + err := cl.Call(ctx, method, req, out) + b.record(node, method, start, err) + return err == nil +} + +func (b *bundle) record(node, rpc string, start time.Time, err error) { + res := collectionResult{Node: node, RPC: rpc, Status: "ok", ElapsedMs: time.Since(start).Milliseconds()} + if err != nil { + res.Status = "error" + res.Error = err.Error() + b.errs = append(b.errs, fmt.Sprintf("[%s] %s: %v", node, rpc, err)) + } + b.results = append(b.results, res) +} + +func (b *bundle) add(name string, data []byte) { + f, err := b.zw.Create(name) + if err != nil { + b.errs = append(b.errs, fmt.Sprintf("zip create %s: %v", name, err)) + return + } + if _, err := f.Write(data); err != nil { + b.errs = append(b.errs, fmt.Sprintf("zip write %s: %v", name, err)) + } +} + +// addJSON re-indents a raw JSON response for readability before writing. +func (b *bundle) addJSON(name string, raw json.RawMessage) { + pretty, err := jsonIndent(raw) + if err != nil { + b.add(name, raw) + return + } + b.add(name, pretty) +} + +func (b *bundle) writeManifest() { + m := manifest{ + BundleCreatedAt: time.Now().UTC().Format(time.RFC3339), + ToolVersion: b.opts.ToolVersion, + RedpandaSQLVersions: b.versions, + NodesAttempted: len(b.versions), + NodesSucceeded: len(b.versions), + CollectionResults: b.results, + RedactionModes: map[string]string{"sql_text": b.opts.SQLTextMode}, + } + out, _ := json.MarshalIndent(m, "", " ") + b.add(bundleRoot+"/manifest.json", out) +} + +func (b *bundle) writeErrors() { + if len(b.errs) == 0 { + return + } + b.add(bundleRoot+"/errors.txt", []byte(strings.Join(b.errs, "\n")+"\n")) +} + +func jsonIndent(raw json.RawMessage) ([]byte, error) { + var v any + if err := json.Unmarshal(raw, &v); err != nil { + return nil, err + } + return json.MarshalIndent(v, "", " ") +} + +// cpuPercentage derives CPU utilization from two resource-usage samples: +// busy CPU-seconds over the wall-clock interval between them. +func cpuPercentage(firstRaw, secondRaw json.RawMessage) (float64, bool) { + var a, c resourceUsageSample + if json.Unmarshal(firstRaw, &a) != nil || json.Unmarshal(secondRaw, &c) != nil { + return 0, false + } + ticksPerSec := atoi(c.ClockTicksPerSec) + wallMs := atoi(c.SampledAtUnixMs) - atoi(a.SampledAtUnixMs) + if ticksPerSec <= 0 || wallMs <= 0 { + return 0, false + } + busyTicks := (atoi(c.CPUUserTicks) + atoi(c.CPUKernelTicks)) - (atoi(a.CPUUserTicks) + atoi(a.CPUKernelTicks)) + busySec := float64(busyTicks) / float64(ticksPerSec) + wallSec := float64(wallMs) / 1000.0 + return busySec / wallSec * 100.0, true +} + +func atoi(s string) int64 { + n, _ := strconv.ParseInt(s, 10, 64) + return n +} + +func nonempty(s, fallback string) string { + if s != "" { + return s + } + return fallback +} diff --git a/src/go/rpk/pkg/cli/sql/debug/bundle/messages.go b/src/go/rpk/pkg/cli/sql/debug/bundle/messages.go new file mode 100644 index 0000000000000..5766ee9c91443 --- /dev/null +++ b/src/go/rpk/pkg/cli/sql/debug/bundle/messages.go @@ -0,0 +1,133 @@ +// 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 + +// Only the fields the collector acts on are modeled here; every other response is +// passed through to its output file verbatim. Wire encoding follows the proto3 +// JSON mapping: field names are lowerCamelCase, 64-bit integers are quoted +// strings, `bytes` are base64 (std) strings, enums are their symbolic names. + +// --- requests --------------------------------------------------------------- + +type getRecentQueriesRequest struct { + IncludeSQLText string `json:"includeSqlText,omitempty"` +} + +type getActiveQueriesRequest struct { + IncludeSQLText string `json:"includeSqlText,omitempty"` +} + +type getLogTailRequest struct { + SinceUnixMs int64 `json:"sinceUnixMs,omitempty"` + SizeLimitBytes uint64 `json:"sizeLimitBytes,omitempty"` +} + +type getHostProbesRequest struct { + IncludeVmstat bool `json:"includeVmstat,omitempty"` +} + +type getCPUProfileRequest struct { + DurationSeconds uint32 `json:"durationSeconds,omitempty"` +} + +// --- responses -------------------------------------------------------------- + +type getVersionResponse struct { + Version string `json:"version"` + CommitSHA string `json:"commitSha"` + ImageTag string `json:"imageTag"` + ImageDigest string `json:"imageDigest"` + Hostname string `json:"hostname"` + NodeID string `json:"nodeId"` +} + +type getConfigResponse struct { + YAML string `json:"yaml"` +} + +type clusterNode struct { + NodeID string `json:"nodeId"` + AdminEndpoint string `json:"adminEndpoint"` + Role string `json:"role"` + MembershipStatus string `json:"membershipStatus"` +} + +type getClusterNodesResponse struct { + Nodes []clusterNode `json:"nodes"` +} + +type getCatalogHeadResponse struct { + CatalogHead []byte `json:"catalogHead"` // base64 in, raw bytes after decode + CatalogHeadJSON string `json:"catalogHeadJson"` // proto decoded to JSON server-side; empty if unparseable +} + +// resourceUsageSample keeps the raw counters as strings (proto3 JSON int64/uint64 +// encoding) and parses them only where arithmetic is needed. +type resourceUsageSample struct { + CPUUserTicks string `json:"cpuUserTicks"` + CPUKernelTicks string `json:"cpuKernelTicks"` + ClockTicksPerSec string `json:"clockTicksPerSec"` + SampledAtUnixMs string `json:"sampledAtUnixMs"` + FreeMemoryBytes string `json:"freeMemoryBytes"` + TotalMemoryBytes string `json:"totalMemoryBytes"` + FreeDiskBytes string `json:"freeDiskBytes"` + TotalDiskBytes string `json:"totalDiskBytes"` +} + +type getLogTailResponse struct { + Content []byte `json:"content"` // base64 in, raw bytes after decode + Truncated bool `json:"truncated"` +} + +type crashReport struct { + TimestampUnixMs string `json:"timestampUnixMs"` + NodeID string `json:"nodeId"` + PID string `json:"pid"` + Filename string `json:"filename"` + Trace string `json:"trace"` +} + +type getCrashReportsResponse struct { + Reports []crashReport `json:"reports"` +} + +type startupLogFile struct { + Filename string `json:"filename"` + MtimeUnixMs string `json:"mtimeUnixMs"` + Content []byte `json:"content"` // base64 in, raw bytes after decode +} + +type getStartupLogResponse struct { + Files []startupLogFile `json:"files"` // newest-first +} + +type getHostProbesResponse struct { + ProcCPUInfo string `json:"procCpuinfo"` + ProcMemInfo string `json:"procMeminfo"` + ProcDiskstats string `json:"procDiskstats"` + ProcLoadavg string `json:"procLoadavg"` + ProcVersion string `json:"procVersion"` + ProcMounts string `json:"procMounts"` + ProcNetSockstat string `json:"procNetSockstat"` + Uname string `json:"uname"` + DF string `json:"df"` + Free string `json:"free"` + Vmstat string `json:"vmstat"` + Top string `json:"top"` + Uptime string `json:"uptime"` + Sysctl string `json:"sysctl"` + ContainerImageTag string `json:"containerImageTag"` + ContainerImageDigst string `json:"containerImageDigest"` + ResolvConf string `json:"resolvConf"` +} + +type getCPUProfileResponse struct { + PprofGzip []byte `json:"pprofGzip"` // base64 in, raw gzip bytes after decode +} diff --git a/src/go/rpk/pkg/cli/sql/debug/debug.go b/src/go/rpk/pkg/cli/sql/debug/debug.go new file mode 100644 index 0000000000000..b8b328aa6ac00 --- /dev/null +++ b/src/go/rpk/pkg/cli/sql/debug/debug.go @@ -0,0 +1,29 @@ +// 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 debug + +import ( + "github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/sql/debug/bundle" + "github.com/redpanda-data/redpanda/src/go/rpk/pkg/config" + "github.com/spf13/afero" + "github.com/spf13/cobra" +) + +// NewCommand returns the `rpk sql debug` command group. +func NewCommand(fs afero.Fs, p *config.Params) *cobra.Command { + cmd := &cobra.Command{ + Use: "debug", + Short: "Debug a Redpanda SQL cluster", + } + cmd.AddCommand( + bundle.NewCommand(fs, p), + ) + return cmd +} diff --git a/src/go/rpk/pkg/cli/sql/sql.go b/src/go/rpk/pkg/cli/sql/sql.go new file mode 100644 index 0000000000000..0bb4d00c17b80 --- /dev/null +++ b/src/go/rpk/pkg/cli/sql/sql.go @@ -0,0 +1,30 @@ +// 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 sql + +import ( + "github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/sql/debug" + "github.com/redpanda-data/redpanda/src/go/rpk/pkg/config" + "github.com/spf13/afero" + "github.com/spf13/cobra" +) + +// NewCommand returns the `rpk sql` command group for interacting with a +// Redpanda SQL cluster. +func NewCommand(fs afero.Fs, p *config.Params) *cobra.Command { + cmd := &cobra.Command{ + Use: "sql", + Short: "Interact with a Redpanda SQL cluster", + } + cmd.AddCommand( + debug.NewCommand(fs, p), + ) + return cmd +}