Skip to content

Commit 3103408

Browse files
Merge pull request #30946 from nguyen-andrew/sl-role-sync-task-pr2
cluster_link: shadow link roles migrator
2 parents 87728ec + e2e4377 commit 3103408

16 files changed

Lines changed: 1699 additions & 11 deletions

src/v/cluster_link/BUILD

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,7 @@ redpanda_cc_library(
8383
"//src/v/cluster:plugin_backend",
8484
"//src/v/cluster:types",
8585
"//src/v/cluster_link/utils:topic_properties_utils",
86+
"//src/v/features",
8687
],
8788
visibility = ["//src/v/cluster_link:__subpackages__"],
8889
deps = [
@@ -109,6 +110,7 @@ redpanda_cc_library(
109110
"//src/v/kafka/data/rpc",
110111
"//src/v/metrics",
111112
"//src/v/model",
113+
"//src/v/security",
112114
"//src/v/ssx:future_util",
113115
"//src/v/ssx:mutex",
114116
"//src/v/ssx:work_queue",
@@ -207,6 +209,35 @@ redpanda_cc_library(
207209
],
208210
)
209211

212+
redpanda_cc_library(
213+
name = "roles_migrator",
214+
srcs = [
215+
"roles_migrator.cc",
216+
],
217+
hdrs = [
218+
"roles_migrator.h",
219+
],
220+
implementation_deps = [
221+
":impl",
222+
":role_reconcile",
223+
"//src/v/kafka/server:security",
224+
"//src/v/security",
225+
],
226+
visibility = ["//src/v/cluster_link:__subpackages__"],
227+
deps = [
228+
":logger",
229+
":role_reconcile",
230+
":utils",
231+
"//src/v/cluster:fwd",
232+
"//src/v/cluster_link/model",
233+
"//src/v/kafka/client:cluster",
234+
"//src/v/kafka/protocol",
235+
"//src/v/kafka/protocol:describe_redpanda_roles",
236+
"//src/v/kafka/server:roles",
237+
"@seastar",
238+
],
239+
)
240+
210241
redpanda_cc_library(
211242
name = "cluster_link",
212243
srcs = [
@@ -219,6 +250,7 @@ redpanda_cc_library(
219250
":group_mirroring_task",
220251
":impl",
221252
":logger",
253+
":roles_migrator",
222254
":security_migrator",
223255
":source_topic_syncer",
224256
"//src/v/cluster:fwd",
@@ -265,6 +297,22 @@ redpanda_cc_library(
265297
],
266298
)
267299

300+
redpanda_cc_library(
301+
name = "role_reconcile",
302+
srcs = [
303+
"role_reconcile.cc",
304+
],
305+
hdrs = [
306+
"role_reconcile.h",
307+
],
308+
visibility = ["//src/v/cluster_link:__subpackages__"],
309+
deps = [
310+
"//src/v/container:chunked_hash_map",
311+
"//src/v/container:chunked_vector",
312+
"//src/v/security",
313+
],
314+
)
315+
268316
redpanda_cc_library(
269317
name = "rpc_service",
270318
srcs = [

src/v/cluster_link/deps.cc

Lines changed: 53 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -14,24 +14,71 @@
1414
#include "cluster/members_table.h"
1515
#include "cluster/security_frontend.h"
1616
#include "cluster_link/utils.h"
17+
#include "features/feature_table.h"
1718
#include "kafka/data/rpc/client.h"
1819
#include "kafka/data/rpc/serde.h"
20+
#include "security/role_store.h"
1921

2022
namespace cluster_link {
2123

2224
namespace {
2325
class security_impl : public security_service {
2426
public:
25-
explicit security_impl(ss::sharded<cluster::security_frontend>* security_fe)
26-
: _security_fe(security_fe) {}
27+
security_impl(
28+
ss::sharded<cluster::security_frontend>* security_fe,
29+
ss::sharded<security::role_store>* role_store,
30+
ss::sharded<features::feature_table>* features)
31+
: _security_fe(security_fe)
32+
, _role_store(role_store)
33+
, _features(features) {}
34+
2735
ss::future<chunked_vector<cluster::errc>> create_acls(
2836
chunked_vector<security::acl_binding> bindings,
2937
::model::timeout_clock::duration timeout) final {
3038
return _security_fe->local().create_acls(std::move(bindings), timeout);
3139
}
3240

41+
ss::future<std::error_code> create_role(
42+
security::role_name name,
43+
security::role role,
44+
::model::timeout_clock::duration timeout) final {
45+
return _security_fe->local().create_role(
46+
std::move(name),
47+
std::move(role),
48+
::model::timeout_clock::now() + timeout);
49+
}
50+
51+
ss::future<std::error_code> update_role(
52+
security::role_name name,
53+
security::role role,
54+
::model::timeout_clock::duration timeout) final {
55+
return _security_fe->local().update_role(
56+
std::move(name),
57+
std::move(role),
58+
::model::timeout_clock::now() + timeout);
59+
}
60+
61+
ss::future<std::error_code> delete_role(
62+
security::role_name name,
63+
::model::timeout_clock::duration timeout) final {
64+
return _security_fe->local().delete_role(
65+
std::move(name), ::model::timeout_clock::now() + timeout);
66+
}
67+
68+
bool rbac_active() const final {
69+
return _features->local().is_active(
70+
features::feature::role_based_access_control);
71+
}
72+
73+
chunked_vector<security::role_with_members> read_shadow_roles(
74+
const std::function<bool(const security::role_name&)>& pred) const final {
75+
return _role_store->local().roles_with_members(pred);
76+
}
77+
3378
private:
3479
ss::sharded<cluster::security_frontend>* _security_fe;
80+
ss::sharded<security::role_store>* _role_store;
81+
ss::sharded<features::feature_table>* _features;
3582
};
3683

3784
class kafka_rpc_client_impl : public kafka_rpc_client_service {
@@ -66,8 +113,10 @@ class members_table_provider_impl : public members_table_provider {
66113
} // namespace
67114

68115
std::unique_ptr<security_service> security_service::make_default(
69-
ss::sharded<cluster::security_frontend>* security_fe) {
70-
return std::make_unique<security_impl>(security_fe);
116+
ss::sharded<cluster::security_frontend>* security_fe,
117+
ss::sharded<security::role_store>* role_store,
118+
ss::sharded<features::feature_table>* features) {
119+
return std::make_unique<security_impl>(security_fe, role_store, features);
71120
}
72121

73122
std::unique_ptr<kafka::client::cluster>

src/v/cluster_link/deps.h

Lines changed: 36 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,16 @@
2020
#include "kafka/data/rpc/fwd.h"
2121
#include "kafka/data/rpc/serde.h"
2222
#include "model/fundamental.h"
23+
#include "security/fwd.h"
24+
#include "security/role.h"
25+
#include "security/types.h"
2326

2427
#include <expected>
28+
#include <functional>
29+
30+
namespace features {
31+
class feature_table;
32+
} // namespace features
2533

2634
namespace cluster_link {
2735

@@ -189,12 +197,38 @@ class security_service {
189197
security_service& operator=(security_service&&) = delete;
190198
virtual ~security_service() = default;
191199

192-
static std::unique_ptr<security_service>
193-
make_default(ss::sharded<cluster::security_frontend>*);
200+
static std::unique_ptr<security_service> make_default(
201+
ss::sharded<cluster::security_frontend>*,
202+
ss::sharded<security::role_store>*,
203+
ss::sharded<features::feature_table>*);
194204

195205
virtual ss::future<chunked_vector<cluster::errc>> create_acls(
196206
chunked_vector<security::acl_binding>,
197207
::model::timeout_clock::duration) = 0;
208+
209+
/// Create a role on the shadow cluster. errc::role_exists if it already
210+
/// exists; errc::feature_disabled if RBAC is not active.
211+
virtual ss::future<std::error_code> create_role(
212+
security::role_name,
213+
security::role,
214+
::model::timeout_clock::duration) = 0;
215+
/// Overwrite a role's membership on the shadow cluster.
216+
/// errc::role_does_not_exist if it is gone; errc::feature_disabled if RBAC
217+
/// is not active.
218+
virtual ss::future<std::error_code> update_role(
219+
security::role_name,
220+
security::role,
221+
::model::timeout_clock::duration) = 0;
222+
/// Delete a role on the shadow cluster. errc::role_does_not_exist if
223+
/// already gone; errc::feature_disabled if RBAC is not active.
224+
virtual ss::future<std::error_code>
225+
delete_role(security::role_name, ::model::timeout_clock::duration) = 0;
226+
/// Whether the role_based_access_control feature is active locally.
227+
virtual bool rbac_active() const = 0;
228+
/// Enumerate shadow-cluster roles (with members) matching the predicate,
229+
/// from the local controller role_store.
230+
virtual chunked_vector<security::role_with_members> read_shadow_roles(
231+
const std::function<bool(const security::role_name&)>&) const = 0;
198232
};
199233

200234
class kafka_rpc_client_service {
Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
/*
2+
* Copyright 2026 Redpanda Data, Inc.
3+
*
4+
* Use of this software is governed by the Business Source License
5+
* included in the file licenses/BSL.md
6+
*
7+
* As of the Change Date specified in that file, in accordance with
8+
* the Business Source License, use of this software will be governed
9+
* by the Apache License, Version 2.0
10+
*/
11+
12+
#include "cluster_link/role_reconcile.h"
13+
14+
#include "container/chunked_hash_map.h"
15+
16+
namespace cluster_link {
17+
18+
role_changes reconcile_roles(
19+
chunked_vector<security::role_with_members> source_selected,
20+
chunked_vector<security::role_with_members> shadow_selected) {
21+
auto to_map = [](chunked_vector<security::role_with_members> roles) {
22+
chunked_hash_map<security::role_name, security::role> m;
23+
m.reserve(roles.size());
24+
for (auto& r : roles) {
25+
m.emplace(std::move(r.name), std::move(r.role));
26+
}
27+
return m;
28+
};
29+
auto source = to_map(std::move(source_selected));
30+
auto shadow = to_map(std::move(shadow_selected));
31+
32+
role_changes changes;
33+
for (auto& [src_name, src_role] : source) {
34+
auto it = shadow.find(src_name);
35+
if (it == shadow.end()) {
36+
changes.to_create.push_back(
37+
{.name = src_name, .role = std::move(src_role)});
38+
} else if (src_role.members() != it->second.members()) {
39+
changes.to_update.push_back(
40+
{.name = src_name, .role = std::move(src_role)});
41+
}
42+
}
43+
44+
for (const auto& [shadow_name, _] : shadow) {
45+
if (!source.contains(shadow_name)) {
46+
changes.to_delete.push_back(shadow_name);
47+
}
48+
}
49+
50+
return changes;
51+
}
52+
53+
} // namespace cluster_link
Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
/*
2+
* Copyright 2026 Redpanda Data, Inc.
3+
*
4+
* Use of this software is governed by the Business Source License
5+
* included in the file licenses/BSL.md
6+
*
7+
* As of the Change Date specified in that file, in accordance with
8+
* the Business Source License, use of this software will be governed
9+
* by the Apache License, Version 2.0
10+
*/
11+
12+
#pragma once
13+
14+
#include "container/chunked_vector.h"
15+
#include "security/role.h"
16+
#include "security/types.h"
17+
18+
namespace cluster_link {
19+
20+
/// The set of role mutations needed to make the shadow cluster's in-scope
21+
/// roles match the source's in-scope roles. Inputs to reconcile_roles are
22+
/// already filtered to the configured selection, so this struct carries no
23+
/// policy.
24+
struct role_changes {
25+
chunked_vector<security::role_with_members> to_create;
26+
chunked_vector<security::role_with_members> to_update;
27+
chunked_vector<security::role_name> to_delete;
28+
};
29+
30+
/// Compute the full-mirror diff between two already-filtered role sets:
31+
/// to_create = source \ shadow (by name)
32+
/// to_update = names in both whose membership differs
33+
/// to_delete = shadow \ source (by name)
34+
role_changes reconcile_roles(
35+
chunked_vector<security::role_with_members> source_selected,
36+
chunked_vector<security::role_with_members> shadow_selected);
37+
38+
} // namespace cluster_link

0 commit comments

Comments
 (0)