Merge pull request #23743 from ztlpn/leadership-pinning-license

Leadership pinning: add to enterprise features

Author: mmaslankaprv

src/v/cluster/controller.cc

@@ -625,6 +625,7 @@ ss::future<> controller::start(
       std::ref(_feature_table),
       std::ref(_connections),
       std::ref(_roles),
+      std::ref(_tp_state),
       _raft0->group());
 
     co_await _health_manager.start_single(

src/v/cluster/feature_manager.cc

@@ -52,6 +52,7 @@ feature_manager::feature_manager(
   ss::sharded<features::feature_table>& table,
   ss::sharded<rpc::connection_cache>& connection_cache,
   ss::sharded<security::role_store>& role_store,
+  ss::sharded<topic_table>& topic_table,
   raft::group_id raft0_group)
   : _stm(stm)
   , _as(as)
@@ -62,6 +63,7 @@ feature_manager::feature_manager(
   , _feature_table(table)
   , _connection_cache(connection_cache)
   , _role_store(role_store)
+  , _topic_table(topic_table)
   , _raft0_group(raft0_group)
   , _barrier_state(
       *config::node().node_id(),
@@ -225,6 +227,18 @@ feature_manager::report_enterprise_features() const {
     auto has_non_default_roles
       = n_roles >= 2
         || (n_roles == 1 && !_role_store.local().contains(security::default_role));
+    auto leadership_pinning_enabled = [&cfg, this]() {
+        if (cfg.default_leaders_preference() != config::leaders_preference{}) {
+            return true;
+        }
+        for (const auto& topic : _topic_table.local().topics_map()) {
+            if (topic.second.get_configuration()
+                  .properties.leaders_preference) {
+                return true;
+            }
+        }
+        return false;
+    };
 
     features::enterprise_feature_report report;
     report.set(
@@ -249,6 +263,9 @@ feature_manager::report_enterprise_features() const {
     report.set(
       features::license_required_feature::datalake_iceberg,
       cfg.iceberg_enabled());
+    report.set(
+      features::license_required_feature::leadership_pinning,
+      leadership_pinning_enabled());
     return report;
 }
 

src/v/cluster/feature_manager.h

@@ -67,6 +67,7 @@ class feature_manager {
       ss::sharded<features::feature_table>& table,
       ss::sharded<rpc::connection_cache>& connection_cache,
       ss::sharded<security::role_store>& role_store,
+      ss::sharded<topic_table>&,
       raft::group_id raft0_group);
 
     /**
@@ -165,6 +166,7 @@ class feature_manager {
     ss::sharded<features::feature_table>& _feature_table;
     ss::sharded<rpc::connection_cache>& _connection_cache;
     ss::sharded<security::role_store>& _role_store;
+    ss::sharded<topic_table>& _topic_table;
     raft::group_id _raft0_group;
 
     version_map _updates;

src/v/features/enterprise_features.cc

@@ -39,6 +39,8 @@ std::ostream& operator<<(std::ostream& os, license_required_feature f) {
         return os << "fips";
     case license_required_feature::datalake_iceberg:
         return os << "datalake_iceberg";
+    case license_required_feature::leadership_pinning:
+        return os << "leadership_pinning";
     }
     __builtin_unreachable();
 }

src/v/features/enterprise_features.h

@@ -29,6 +29,7 @@ enum class license_required_feature {
     rbac,
     fips,
     datalake_iceberg,
+    leadership_pinning,
 };
 
 std::ostream& operator<<(std::ostream&, license_required_feature);

tests/rptest/tests/enterprise_features_license_test.py

@@ -3,6 +3,7 @@
 from enum import IntEnum
 
 from rptest.utils.rpenv import sample_license
+from rptest.clients.rpk import RpkTool
 from rptest.services.admin import Admin, EnterpriseLicenseStatus, RolesList, RoleDescription
 from rptest.services.redpanda import RESTART_LOG_ALLOW_LIST, SecurityConfig, SchemaRegistryConfig
 from rptest.tests.redpanda_test import RedpandaTest
@@ -38,6 +39,7 @@ class Features(IntEnum):
     rbac = 7
     fips = 8
     datalake_iceberg = 9
+    leadership_pinning = 10
 
 
 SKIP_FEATURES = [
@@ -191,6 +193,12 @@ def has_role(r: str):
         elif feature == Features.datalake_iceberg:
             self.redpanda.set_cluster_config({'iceberg_enabled': 'true'},
                                              expect_restart=True)
+        elif feature == Features.leadership_pinning:
+            RpkTool(self.redpanda).create_topic(
+                "foo",
+                partitions=1,
+                replicas=1,
+                config={"redpanda.leaders.preference": "racks:rack1"})
         else:
             assert False, f"Unexpected feature={feature}"