Merge branch 'v26.1.x' into backport-pr-29970-v26.1.x-827

Author: piyushredpanda

bazel/pbgen/main.go

@@ -635,6 +635,7 @@ func (g *headerGenerator) generateMessage(msg protoreflect.MessageDescriptor, w
 	}
 	w.Println()
 	w.Printf("std::string_view full_name() const override { return %q; }\n", msg.FullName())
+	w.Printf("static constexpr size_t field_count = %d;\n", msg.Fields().Len())
 	w.Println("// Convert a field path into a path of field numbers.")
 	w.Println("static bool convert_field_path_to_numbers(std::span<std::string_view> field_path, std::vector<int32_t>* out);")
 	w.Println("// Convert a field path into a path of field numbers.")

proto/redpanda/core/admin/internal/cloud_topics/v1/metastore.proto

@@ -88,6 +88,27 @@ service MetastoreService {
             authz: SUPERUSER
         };
     }
+
+    // List cloud topics from the topic table, returning their names, UUIDs,
+    // and partition counts. Supports pagination by topic name.
+    rpc ListCloudTopics(ListCloudTopicsRequest)
+        returns (ListCloudTopicsResponse) {
+        option (pbgen.rpc) = {
+            authz: SUPERUSER
+        };
+    }
+
+    // Validate metastore state for a single partition. Checks invariants
+    // such as extent contiguity, offset bounds, and term ordering.
+    // Supports pagination via max_extents and resume_after_offset to
+    // bound the work per call. Cross-page contiguity is enforced by
+    // checking the boundary between consecutive pages.
+    rpc ValidatePartition(ValidatePartitionRequest)
+        returns (ValidatePartitionResponse) {
+        option (pbgen.rpc) = {
+            authz: SUPERUSER
+        };
+    }
 }
 
 // GetOffsetsRequest is the request for looking up the offsets in the metastore
@@ -215,6 +236,7 @@ message MetadataValue {
     int64 next_offset = 2;
     int64 compaction_epoch = 3;
     uint64 size = 4;
+    uint64 num_extents = 5;
 }
 
 // Value for an extent row.
@@ -324,3 +346,96 @@ message ReadRowsResponse {
     // Raw key for the next page. Empty if no more rows.
     string next_key = 2;
 }
+
+// ValidatePartitionRequest validates metastore invariants for a single
+// partition. Supports paginated validation via max_extents and
+// resume_at_offset.
+message ValidatePartitionRequest {
+    // The topic UUID to validate.
+    string topic_id = 1;
+    // The partition ID to validate.
+    int32 partition_id = 2;
+    // If true, verify each extent's object_id exists in the metastore
+    // object table and is not a preregistration.
+    bool check_object_metadata = 3;
+    // If true, verify each extent's L1 object exists in cloud storage
+    // via HEAD request.
+    bool check_object_storage = 4;
+    // Resume validation at this base_offset (inclusive). On the next
+    // page, the validator re-reads the extent at this offset to verify
+    // cross-page contiguity, then continues forward. The re-read extent
+    // is not counted toward max_extents. Omit to start from the
+    // beginning of the partition.
+    optional int64 resume_at_offset = 5;
+    // Max extents to validate per call (excluding the re-read extent on
+    // continuation pages). 0 means validate all remaining extents.
+    uint32 max_extents = 6;
+}
+
+// ValidatePartitionResponse contains the results of partition validation.
+message ValidatePartitionResponse {
+    // Anomalies found during validation. Empty if the partition is valid.
+    repeated MetastoreAnomaly anomalies = 1;
+    // The base_offset of the last extent validated. If absent, all
+    // extents have been validated. Pass this value as resume_at_offset
+    // in the next call to continue validation.
+    optional int64 resume_at_offset = 2;
+    // Number of extents validated in this call (excluding the re-read
+    // extent on continuation pages).
+    uint32 extents_validated = 3;
+}
+
+// Type of anomaly found during metastore validation.
+enum AnomalyType {
+    ANOMALY_TYPE_UNSPECIFIED = 0;
+    ANOMALY_TYPE_EXTENT_OVERLAP = 1;
+    ANOMALY_TYPE_EXTENT_GAP = 2;
+    ANOMALY_TYPE_NEXT_OFFSET_MISMATCH = 3;
+    ANOMALY_TYPE_START_OFFSET_MISMATCH = 4;
+    ANOMALY_TYPE_OBJECT_NOT_FOUND = 5;
+    ANOMALY_TYPE_OBJECT_PREREGISTERED = 6;
+    ANOMALY_TYPE_OBJECT_NOT_IN_STORAGE = 7;
+    ANOMALY_TYPE_COMPACTION_RANGE_BELOW_START = 8;
+    ANOMALY_TYPE_COMPACTION_RANGE_ABOVE_NEXT = 9;
+    ANOMALY_TYPE_COMPACTION_TOMBSTONE_OVERLAP = 10;
+    ANOMALY_TYPE_COMPACTION_TOMBSTONE_OUTSIDE_CLEANED = 11;
+    ANOMALY_TYPE_TERM_ORDERING = 12;
+    ANOMALY_TYPE_COMPACTION_STATE_UNEXPECTED = 13;
+}
+
+// A single anomaly found during metastore validation.
+message MetastoreAnomaly {
+    // The type of anomaly.
+    AnomalyType anomaly_type = 1;
+    // Human-readable description of the anomaly.
+    string description = 2;
+}
+
+// ListCloudTopicsRequest lists cloud topics from the cluster's topic table
+// (not the metastore). Hosted here for convenience since the metastore
+// admin service already has topic_table access.
+message ListCloudTopicsRequest {
+    // Optional: resume listing after this topic name (exclusive).
+    // Omit to start from the beginning.
+    string after_topic_name = 1;
+    // Maximum number of topics to return. 0 uses a default of 100.
+    uint32 max_topics = 2;
+}
+
+// ListCloudTopicsResponse contains cloud topics with their metadata.
+message ListCloudTopicsResponse {
+    repeated CloudTopicInfo topics = 1;
+    // If true, more topics are available. Call again with after_topic_name
+    // set to the last topic's name to continue.
+    bool has_more = 2;
+}
+
+// Information about a single cloud topic.
+message CloudTopicInfo {
+    // The Kafka topic name.
+    string topic_name = 1;
+    // The topic UUID.
+    string topic_id = 2;
+    // Number of partitions.
+    int32 partition_count = 3;
+}

src/go/rpk/pkg/cli/cloud/auth/BUILD

@@ -9,6 +9,7 @@ go_library(
         "edit.go",
         "list.go",
         "rename.go",
+        "token.go",
         "use.go",
     ],
     importpath = "github.com/redpanda-data/redpanda/src/go/rpk/pkg/cli/cloud/auth",

src/go/rpk/pkg/cli/cloud/auth/auth.go

@@ -38,6 +38,7 @@ only use a single SSO based login.
 		newEditCommand(fs, p),
 		newListCommand(fs, p),
 		newRenameToCommand(fs, p),
+		newTokenCommand(fs, p),
 		newUseCommand(fs, p),
 	)
 

src/go/rpk/pkg/cli/cloud/auth/token.go

@@ -0,0 +1,50 @@
+// Copyright 2025 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 auth
+
+import (
+	"fmt"
+
+	"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 newTokenCommand(fs afero.Fs, p *config.Params) *cobra.Command {
+	return &cobra.Command{
+		Use:   "token",
+		Short: "Print the cloud auth token of the current profile",
+		Long: `Print the cloud auth token of the current profile.
+
+This command prints the auth token for the currently selected cloud
+authentication to stdout. This is useful for piping the token into other
+commands or scripts that need to authenticate with Redpanda Cloud.`,
+		Args: cobra.ExactArgs(0),
+		Run: func(_ *cobra.Command, _ []string) {
+			cfg, err := p.Load(fs)
+			out.MaybeDie(err, "rpk unable to load config: %v", err)
+
+			y, ok := cfg.ActualRpkYaml()
+			if !ok {
+				out.Die("rpk.yaml file does not exist")
+			}
+
+			a := y.CurrentAuth()
+			if a == nil {
+				out.Die("no current cloud auth selected, use 'rpk cloud login' to log in")
+			}
+			if a.AuthToken == "" {
+				out.Die("current auth has no token, use 'rpk cloud login' to log in")
+			}
+			fmt.Print(a.AuthToken)
+		},
+	}
+}

src/go/rpk/pkg/cli/security/role/BUILD

@@ -1,4 +1,4 @@
-load("@rules_go//go:def.bzl", "go_library")
+load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "role",
@@ -19,8 +19,6 @@ go_library(
         "//src/go/rpk/pkg/kafka",
         "//src/go/rpk/pkg/out",
         "//src/go/rpk/pkg/publicapi",
-        "//src/go/rpk/pkg/redpanda",
-        "@build_buf_gen_go_redpandadata_core_protocolbuffers_go//redpanda/core/admin/v2:admin",
         "@build_buf_gen_go_redpandadata_dataplane_protocolbuffers_go//redpanda/api/dataplane/v1:dataplane",
         "@com_connectrpc_connect//:connect",
         "@com_github_redpanda_data_common_go_rpadmin//:rpadmin",
@@ -30,3 +28,10 @@ go_library(
         "@com_github_twmb_types//:types",
     ],
 )
+
+go_test(
+    name = "role_test",
+    srcs = ["role_test.go"],
+    embed = [":role"],
+    deps = ["@com_github_stretchr_testify//require"],
+)

src/go/rpk/pkg/cli/security/role/assign.go

@@ -12,7 +12,6 @@ package role
 import (
 	"fmt"
 
-	adminv2 "buf.build/gen/go/redpandadata/core/protocolbuffers/go/redpanda/core/admin/v2"
 	dataplanev1 "buf.build/gen/go/redpandadata/dataplane/protocolbuffers/go/redpanda/api/dataplane/v1"
 	"connectrpc.com/connect"
 	"github.com/redpanda-data/common-go/rpadmin"
@@ -26,21 +25,15 @@ import (
 
 func assignCommand(fs afero.Fs, p *config.Params) *cobra.Command {
 	var principals []string
-	var groups []string
 	cmd := &cobra.Command{
-		Use:     "assign [ROLE] --principal [PRINCIPALS...] --group [GROUPS...]",
+		Use:     "assign [ROLE] --principal [PRINCIPALS...]",
 		Aliases: []string{"add"},
-		Short:   "Assign a Redpanda role to a principal or group",
-		Long: `Assign a Redpanda role to a principal or group.
+		Short:   "Assign a Redpanda role to a principal",
+		Long: `Assign a Redpanda role to a principal.
 
 The '--principal' flag accepts principals with the format
 '<PrincipalPrefix>:<Principal>'. If 'PrincipalPrefix' is not provided, then
 defaults to 'User:'.
-
-The '--group' flag assigns the role to an identity provider group, granting all members
-of that group the permissions associated with the role. Group assignments are only
-supported on local (non-cloud) clusters running
-Redpanda ` + minGroupVersion.String() + ` or later.
 `,
 		Example: `
 Assign role "redpanda-admin" to user "red"
@@ -49,11 +42,8 @@ Assign role "redpanda-admin" to user "red"
 Assign role "redpanda-admin" to users "red" and "panda"
   rpk security role assign redpanda-admin --principal red,panda
 
-Assign role "data-reader" to group "engineering"
-  rpk security role assign data-reader --group engineering
-
-Assign role "data-reader" to both a user and a group
-  rpk security role assign data-reader --principal alice --group engineering
+Assign role "redpanda-admin" to group "pandas"
+  rpk security role assign redpanda-admin --principal Group:pandas
 `,
 		Args: cobra.ExactArgs(1),
 		Run: func(cmd *cobra.Command, args []string) {
@@ -89,28 +79,6 @@ Assign role "data-reader" to both a user and a group
 				}
 			}
 
-			// Handle groups (local-only).
-			if len(groups) > 0 {
-				if prof.CheckFromCloud() {
-					out.Die("--group is not supported for cloud clusters")
-				}
-				cl, err := adminapi.NewClient(cmd.Context(), fs, prof)
-				out.MaybeDie(err, "unable to initialize admin api client: %v", err)
-
-				if !adminapi.HasMinimumVersion(cmd.Context(), cl, minGroupVersion) {
-					out.Die("--group requires Redpanda version %s or later", minGroupVersion.String())
-				}
-				members := make([]*adminv2.RoleMember, len(groups))
-				for i, g := range groups {
-					members[i] = &adminv2.RoleMember{Member: &adminv2.RoleMember_Group{Group: &adminv2.RoleGroup{Name: g}}}
-				}
-				_, err = cl.SecurityService().AddRoleMembers(cmd.Context(), connect.NewRequest(&adminv2.AddRoleMembersRequest{
-					RoleName: roleName,
-					Members:  members,
-				}))
-				out.MaybeDie(err, "unable to assign group(s) to role %q: %v", roleName, err)
-			}
-
 			// Output principals.
 			if len(toAdd) > 0 {
 				if isText, _, s, err := f.Format(toAdd); !isText {
@@ -124,16 +92,10 @@ Assign role "data-reader" to both a user and a group
 					tw.PrintStructFields(m)
 				}
 			}
-
-			// Output groups.
-			if len(groups) > 0 {
-				fmt.Printf("Successfully assigned role %q to group(s) %v\n", roleName, groups)
-			}
 		},
 	}
 
 	cmd.Flags().StringSliceVar(&principals, "principal", nil, "Principal to assign the role to (repeatable)")
-	cmd.Flags().StringSliceVar(&groups, "group", nil, "group to assign the role to (repeatable)")
-	cmd.MarkFlagsOneRequired("principal", "group")
+	cmd.MarkFlagRequired("principal")
 	return cmd
 }

src/go/rpk/pkg/cli/security/role/role.go

@@ -16,18 +16,16 @@ import (
 	"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/redpanda"
 	"github.com/spf13/afero"
 	"github.com/spf13/cobra"
 )
 
 const (
-	rolePrefix = "RedpandaRole:"
-	userPrefix = "User:"
+	rolePrefix  = "RedpandaRole:"
+	userPrefix  = "User:"
+	groupPrefix = "Group:"
 )
 
-var minGroupVersion = redpanda.Version{Major: 26, Feature: 1}
-
 func NewCommand(fs afero.Fs, p *config.Params) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:     "role",
@@ -51,9 +49,13 @@ func NewCommand(fs afero.Fs, p *config.Params) *cobra.Command {
 // parsePrincipal returns the prefix, and principal. If no prefix is present,
 // returns 'User'.
 func parsePrincipal(p string) (principalType string, name string) {
-	if strings.HasPrefix(p, userPrefix) {
-		return "User", strings.TrimPrefix(p, userPrefix)
+	if s, ok := strings.CutPrefix(p, userPrefix); ok {
+		return "User", s
+	}
+	if s, ok := strings.CutPrefix(p, groupPrefix); ok {
+		return "Group", s
 	}
+
 	return "User", p
 }