Support limits the max namespaces per tenant (apache#8267)

Fixes apache#8224 ### Motivation Support limits the max namespaces per tenant of the Pulsar cluster. When the namespaces reach the max namespaces of the tenant, the broker should reject the new namespace request. ### Modifications Add maxNamespacePerTenant=0 in the broker.conf and don't limit by default. ### Verifying this change AdminApiTest2#testMaxNamespacePerTenant
vikingcodes · Oct 17, 2020 · 9c821cf · 9c821cf
1 parent 682ceb4
commit 9c821cf
Show file tree

Hide file tree

Showing 5 changed files with 53 additions and 1 deletion.
diff --git a/conf/broker.conf b/conf/broker.conf
@@ -205,6 +205,10 @@ brokerDeduplicationProducerInactivityTimeoutMinutes=360
 # value will be used as the default
 defaultNumberOfNamespaceBundles=4
 
+# The maximum number of namespaces that each tenant can create
+# This configuration is not precise control, in a concurrent scenario, the threshold will be exceeded
+maxNamespacesPerTenant=0
+
 # Enable check for minimum allowed client library version
 clientLibraryVersionCheckEnabled=false
 

diff --git a/deployment/terraform-ansible/templates/broker.conf b/deployment/terraform-ansible/templates/broker.conf
@@ -190,6 +190,10 @@ brokerDeduplicationProducerInactivityTimeoutMinutes=360
 # value will be used as the default
 defaultNumberOfNamespaceBundles=4
 
+# The maximum number of namespaces that each tenant can create
+# This configuration is not precise control, in a concurrent scenario, the threshold will be exceeded
+maxNamespacesPerTenant=0
+
 # Enable check for minimum allowed client library version
 clientLibraryVersionCheckEnabled=false
 

diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -430,6 +430,13 @@ public class ServiceConfiguration implements PulsarConfiguration {
         doc = "When a namespace is created without specifying the number of bundle, this"
             + " value will be used as the default")
     private int defaultNumberOfNamespaceBundles = 4;
+
+    @FieldContext(
+        category = CATEGORY_POLICIES, 
+        dynamic = true,
+        doc = "The maximum number of namespaces that each tenant can create."
+            + "This configuration is not precise control, in a concurrent scenario, the threshold will be exceeded")
+    private int maxNamespacesPerTenant = 0;
 
     @FieldContext(
         category = CATEGORY_SERVER,

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -49,7 +49,6 @@
 import javax.ws.rs.core.UriBuilder;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.broker.PulsarServerException;
-import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.admin.AdminResource;
 import org.apache.pulsar.broker.authorization.AuthorizationService;
 import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionBusyException;
@@ -133,6 +132,15 @@ protected void internalCreateNamespace(Policies policies) {
         validatePolicies(namespaceName, policies);
 
         try {
+            int maxNamespacesPerTenant = pulsar().getConfiguration().getMaxNamespacesPerTenant();
+            //no distributed locks are added here.In a concurrent scenario, the threshold will be exceeded.
+            if (maxNamespacesPerTenant > 0) {
+                List<String> namespaces = getListOfNamespaces(namespaceName.getTenant());
+                if (namespaces != null && namespaces.size() > maxNamespacesPerTenant) {
+                    throw new RestException(Status.PRECONDITION_FAILED,
+                            "Exceed the maximum number of namespace in tenant :" + namespaceName.getTenant());
+                }
+            }
             policiesCache().invalidate(path(POLICIES, namespaceName.toString()));
 
             zkCreateOptimistic(path(POLICIES, namespaceName.toString()), jsonMapper().writeValueAsBytes(policies));

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
@@ -1294,4 +1294,33 @@ public void testUpdateClusterWithProxyUrl() throws Exception {
         admin.clusters().updateCluster(clusterName, cluster);
         Assert.assertEquals(admin.clusters().getCluster(clusterName), cluster);
     }
+
+    @Test
+    public void testMaxNamespacesPerTenant() throws Exception {
+        super.internalCleanup();
+        conf.setMaxNamespacesPerTenant(2);
+        super.internalSetup();
+        admin.clusters().createCluster("test", new ClusterData(brokerUrl.toString()));
+        TenantInfo tenantInfo = new TenantInfo(Sets.newHashSet("role1", "role2"), Sets.newHashSet("test"));
+        admin.tenants().createTenant("testTenant", tenantInfo);
+        admin.namespaces().createNamespace("testTenant/ns1", Sets.newHashSet("test"));
+        admin.namespaces().createNamespace("testTenant/ns2", Sets.newHashSet("test"));
+        try {
+            admin.namespaces().createNamespace("testTenant/ns3", Sets.newHashSet("test"));
+        } catch (PulsarAdminException e) {
+            Assert.assertEquals(e.getStatusCode(), 412);
+            Assert.assertEquals(e.getHttpError(), "Exceed the maximum number of namespace in tenant :testTenant");
+        }
+
+        //unlimited
+        super.internalCleanup();
+        conf.setMaxNamespacesPerTenant(0);
+        super.internalSetup();
+        admin.clusters().createCluster("test", new ClusterData(brokerUrl.toString()));
+        admin.tenants().createTenant("testTenant", tenantInfo);
+        for (int i = 0; i < 10; i++) {
+            admin.namespaces().createNamespace("testTenant/ns-" + i, Sets.newHashSet("test"));
+        }
+
+    }
 }