Repository: kafka-ops/kafka-topology-builder (java)

Issue:
Service Accounts - Limit scope of add/delete to a prefix.
**Is your feature request related to a problem? Please describe.**
Where a cluster is shared between org units, each org unit is managing their topics and users within a single cluster, by convention/internal rules, each org uses a prefix for everything, topic names, users to avoid any overlap.

The issue is that on a run of the topology builder, when deleting a service account it has no ability to filter for a prefix, where it should ONLY delete service accounts with said prefix.

**Describe the solution you'd like**
Ability to have a prefix/regex pattern defined, that ensures the tool only deletes service accounts matching a certain regex pattern/prefix.

**Describe alternatives you've considered**
Not use KTB and manage this all seperately.

**Additional context**
Currently means we cannot use KTB for managing service accounts in a shared cluster where different org units manage their own users/topics.


Pull request:
Add support for a prefix list to filter which principals are allowed to be managed
* **Please check if the PR fulfills these requirements**
- [x] The commit messages are descriptive
- [x] Tests for the changes have been added (for bug fixes / features)
- [x] Docs have been added / updated (for bug fixes / features)
- [x] An issue has been created for the pull requests. Some issues might require previous discussion.

closes #178 

Patch:
diff --git a/docs/config-values.rst b/docs/config-values.rst
index 48d12fd..ccc134a 100644
--- a/docs/config-values.rst
+++ b/docs/config-values.rst
@@ -1,3 +1,4 @@
+.. _config:
 Important configuration values
 *******************************
 
@@ -150,4 +151,19 @@ default in 2.0 and in 3.0 the management using the target cluster will be comple
 
 An example to use local topic management state will look like this:
 ::
-    topology.state.topics.cluster.enabled=false
\ No newline at end of file
+    topology.state.topics.cluster.enabled=false
+
+Control allowed Service accounts to be managed by KTB
+-----------
+
+This property is used to control which Service Accounts are allowed to be managed by the KTB, this variable contains a list of allowed prefixes.
+
+**Property**: *topology.service.accounts.managed.prefixes*
+**Default value**: "[]"
+
+An example configuration might look like this:
+::
+    topology.service.accounts.managed.prefixes.0=User:AService
+    topology.service.accounts.managed.prefixes.1=User:BService
+
+If this prefix list is used, only service accounts that match the prefix will be ever processed, anything else will be ignored.
\ No newline at end of file
diff --git a/docs/futures/what-principal-management.rst b/docs/futures/what-principal-management.rst
index 0921b9e..d2cbcb7 100644
--- a/docs/futures/what-principal-management.rst
+++ b/docs/futures/what-principal-management.rst
@@ -23,6 +23,9 @@ principals are present in the cluster.
 This process is done by parsing the topology description and extracting the principals, once done the tool will calculate which principals are
 required to be create and which ones are no longer necessary and can be disabled. For sure always if the related delete option is enabled.
 
+If desired by organisational purposes a user can decide to filter witch Service Account principals can be managed, this is done using the
+*topology.service.accounts.managed.prefixes* configuration setting. Check :ref:`config` for details.
+
 Principals Manager Providers
 -----------
 
diff --git a/src/main/java/com/purbon/kafka/topology/PrincipalManager.java b/src/main/java/com/purbon/kafka/topology/PrincipalManager.java
index f20bba9..65785ae 100644
--- a/src/main/java/com/purbon/kafka/topology/PrincipalManager.java
+++ b/src/main/java/com/purbon/kafka/topology/PrincipalManager.java
@@ -19,6 +19,7 @@ import org.apache.logging.log4j.Logger;
 public class PrincipalManager {
 
   private static final Logger LOGGER = LogManager.getLogger(PrincipalManager.class);
+  private final List<String> managedPrefixes;
 
   private PrincipalProvider provider;
 
@@ -27,6 +28,7 @@ public class PrincipalManager {
   public PrincipalManager(PrincipalProvider provider, TopologyBuilderConfig config) {
     this.provider = provider;
     this.config = config;
+    this.managedPrefixes = config.getServiceAccountManagedPrefixes();
   }
 
   public void apply(Topology topology, ExecutionPlan plan) throws IOException {
@@ -77,9 +79,20 @@ public class PrincipalManager {
             ? provider.listServiceAccounts()
             : plan.getServiceAccounts();
     return accounts.stream()
+        .filter(serviceAccount -> matchesPrefixList(serviceAccount.getName()))
         .collect(Collectors.toMap(ServiceAccount::getName, serviceAccount -> serviceAccount));
   }
 
+  private boolean matchesPrefixList(String principal) {
+    boolean matches =
+        managedPrefixes.size() == 0
+            || managedPrefixes.stream().anyMatch(principal::startsWith);
+    LOGGER.debug(
+        String.format(
+            "Principal %s matches %s with $s", principal, matches, managedPrefixes));
+    return matches;
+  }
+
   private List<String> parseListOfPrincipals(Topology topology) {
     return topology.getProjects().stream()
         .flatMap(
@@ -97,6 +110,7 @@ public class PrincipalManager {
               return users.stream();
             })
         .map(User::getPrincipal)
+        .filter(this::matchesPrefixList)
         .collect(Collectors.toList());
   }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
index 9661927..696e97b 100644
--- a/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
+++ b/src/main/java/com/purbon/kafka/topology/TopologyBuilderConfig.java
@@ -75,6 +75,9 @@ public class TopologyBuilderConfig {
   public static final String TOPOLOGY_TOPIC_STATE_FROM_CLUSTER =
       "topology.state.topics.cluster.enabled";
 
+  static final String SERVICE_ACCOUNT_MANAGED_PREFIXES =
+      "topology.service.accounts.managed.prefixes";
+
   private final Map<String, String> cliParams;
   private Config config;
 
@@ -232,6 +235,12 @@ public class TopologyBuilderConfig {
         .collect(Collectors.toList());
   }
 
+  public List<String> getServiceAccountManagedPrefixes() {
+    return config.getStringList(SERVICE_ACCOUNT_MANAGED_PREFIXES).stream()
+        .map(String::trim)
+        .collect(Collectors.toList());
+  }
+
   public String getConfluentSchemaRegistryUrl() {
     return config.getString(CONFLUENT_SCHEMA_REGISTRY_URL_CONFIG);
   }
diff --git a/src/main/resources/reference.conf b/src/main/resources/reference.conf
index 2b0716f..1efb73c 100644
--- a/src/main/resources/reference.conf
+++ b/src/main/resources/reference.conf
@@ -46,6 +46,11 @@ topology {
         cluster.enabled = true
      }
   }
+  service {
+    accounts {
+      managed.prefixes = []
+    }
+  }
 }
 
 schema {


Test patch:
diff --git a/src/test/java/com/purbon/kafka/topology/PrincipalManagerTest.java b/src/test/java/com/purbon/kafka/topology/PrincipalManagerTest.java
index e8c5ca3..403b620 100644
--- a/src/test/java/com/purbon/kafka/topology/PrincipalManagerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/PrincipalManagerTest.java
@@ -2,6 +2,7 @@ package com.purbon.kafka.topology;
 
 import static com.purbon.kafka.topology.BuilderCLI.ALLOW_DELETE_OPTION;
 import static com.purbon.kafka.topology.BuilderCLI.BROKERS_OPTION;
+import static com.purbon.kafka.topology.TopologyBuilderConfig.SERVICE_ACCOUNT_MANAGED_PREFIXES;
 import static com.purbon.kafka.topology.TopologyBuilderConfig.TOPOLOGY_EXPERIMENTAL_ENABLED_CONFIG;
 import static com.purbon.kafka.topology.TopologyBuilderConfig.TOPOLOGY_TOPIC_STATE_FROM_CLUSTER;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -189,4 +190,35 @@ public class PrincipalManagerTest {
 
     verify(mockPlan, times(0)).add(any(Action.class));
   }
+
+  @Test
+  public void testToProcessOnlySelectedPrincipals() throws IOException {
+
+    props.put(SERVICE_ACCOUNT_MANAGED_PREFIXES, Collections.singletonList("pro"));
+
+    config = new TopologyBuilderConfig(cliOps, props);
+    principalManager = new PrincipalManager(provider, config);
+
+    Topology topology = new TopologyImpl();
+    topology.setContext("context");
+    Project project = new ProjectImpl("foo");
+    project.setConsumers(Collections.singletonList(new Consumer("consumer")));
+    project.setProducers(Collections.singletonList(new Producer("producer")));
+    topology.addProject(project);
+
+    doNothing().when(provider).configure();
+
+    doReturn(new ServiceAccount(123, "consumer", "Managed by KTB"))
+        .when(provider)
+        .createServiceAccount(eq("consumer"), eq("Managed by KTB"));
+
+    doReturn(new ServiceAccount(124, "producer", "Managed by KTB"))
+        .when(provider)
+        .createServiceAccount(eq("producer"), eq("Managed by KTB"));
+
+    principalManager.apply(topology, plan);
+    plan.run();
+
+    assertThat(plan.getServiceAccounts()).hasSize(1);
+  }
 }