From fc9f1632283c15a22d174ec82613c11adc635945 Mon Sep 17 00:00:00 2001 From: Jia Zhai Date: Wed, 16 Dec 2020 23:52:07 +0800 Subject: [PATCH] make namespaces policy update take effect on time --- .../pulsar/broker/ServiceConfiguration.java | 7 ++ .../broker/admin/impl/ClustersBase.java | 113 ++++++++++++++++-- .../broker/loadbalance/LoadManager.java | 8 ++ .../loadbalance/ModularLoadManager.java | 7 ++ .../impl/ModularLoadManagerImpl.java | 9 +- .../impl/ModularLoadManagerWrapper.java | 5 + .../pulsar/broker/admin/AdminApiTest2.java | 57 +++++++++ .../apache/pulsar/broker/admin/AdminTest.java | 3 +- 8 files changed, 199 insertions(+), 10 deletions(-) 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 index 97799cde6aa9a..d065c855de65a 100644 --- 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 @@ -882,6 +882,13 @@ public class ServiceConfiguration implements PulsarConfiguration { doc = "List of interceptors for entry metadata.") private Set brokerEntryMetadataInterceptors = new HashSet<>(); + @FieldContext( + category = CATEGORY_SERVER, + doc = "Enable namespaceIsolation policy update take effect ontime or not," + + " if set to ture, then the related namespaces will be unloaded after reset policy to make it take effect." + ) + private boolean enableNamespaceIsolationUpdateOnTime = false; + /***** --- TLS --- ****/ @FieldContext( category = CATEGORY_TLS, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java index 5e9dfbf3f166b..d2714ec77ce25 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java @@ -37,6 +37,8 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import javax.ws.rs.DELETE; import javax.ws.rs.GET; @@ -44,12 +46,16 @@ import javax.ws.rs.PUT; import javax.ws.rs.Path; import javax.ws.rs.PathParam; +import javax.ws.rs.container.AsyncResponse; +import javax.ws.rs.container.Suspended; import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.Response; import javax.ws.rs.core.Response.Status; import org.apache.bookkeeper.util.ZkUtils; import org.apache.pulsar.broker.admin.AdminResource; import org.apache.pulsar.broker.cache.ConfigurationCacheService; import org.apache.pulsar.broker.web.RestException; +import org.apache.pulsar.client.admin.Namespaces; import org.apache.pulsar.common.naming.Constants; import org.apache.pulsar.common.naming.NamedEntity; import org.apache.pulsar.common.policies.data.BrokerNamespaceIsolationData; @@ -58,6 +64,7 @@ import org.apache.pulsar.common.policies.data.NamespaceIsolationData; import org.apache.pulsar.common.policies.impl.NamespaceIsolationPolicies; import org.apache.pulsar.common.policies.impl.NamespaceIsolationPolicyImpl; +import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.common.util.ObjectMapperFactory; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; @@ -675,6 +682,7 @@ public BrokerNamespaceIsolationData getBrokerWithNamespaceIsolationPolicy( @ApiResponse(code = 500, message = "Internal server error.") }) public void setNamespaceIsolationPolicy( + @Suspended final AsyncResponse asyncResponse, @ApiParam( value = "The cluster name", required = true @@ -690,14 +698,16 @@ public void setNamespaceIsolationPolicy( required = true ) NamespaceIsolationData policyData - ) throws Exception { + ) { validateSuperUserAccess(); validateClusterExists(cluster); validatePoliciesReadOnlyAccess(); + String jsonInput = null; try { // validate the policy data before creating the node policyData.validate(); + jsonInput = ObjectMapperFactory.create().writeValueAsString(policyData); String nsIsolationPolicyPath = path("clusters", cluster, NAMESPACE_ISOLATION_POLICIES); NamespaceIsolationPolicies nsIsolationPolicies = namespaceIsolationPoliciesCache() @@ -715,24 +725,113 @@ public void setNamespaceIsolationPolicy( -1); // make sure that the cache content will be refreshed for the next read access namespaceIsolationPoliciesCache().invalidate(nsIsolationPolicyPath); + + // whether or not make the isolation update on time. + if (pulsar().getConfiguration().isEnableNamespaceIsolationUpdateOnTime()) { + filterAndUnloadMatchedNameSpaces(asyncResponse, policyData); + } else { + asyncResponse.resume(Response.noContent().build()); + return; + } } catch (IllegalArgumentException iae) { log.info("[{}] Failed to update clusters/{}/namespaceIsolationPolicies/{}. Input data is invalid", clientAppId(), cluster, policyName, iae); - String jsonInput = ObjectMapperFactory.create().writeValueAsString(policyData); - throw new RestException(Status.BAD_REQUEST, - "Invalid format of input policy data. policy: " + policyName + "; data: " + jsonInput); + asyncResponse.resume(new RestException(Status.BAD_REQUEST, + "Invalid format of input policy data. policy: " + policyName + "; data: " + jsonInput)); } catch (KeeperException.NoNodeException nne) { log.warn("[{}] Failed to update clusters/{}/namespaceIsolationPolicies: Does not exist", clientAppId(), cluster); - throw new RestException(Status.NOT_FOUND, - "NamespaceIsolationPolicies for cluster " + cluster + " does not exist"); + asyncResponse.resume(new RestException(Status.NOT_FOUND, + "NamespaceIsolationPolicies for cluster " + cluster + " does not exist")); } catch (Exception e) { log.error("[{}] Failed to update clusters/{}/namespaceIsolationPolicies/{}", clientAppId(), cluster, policyName, e); - throw new RestException(e); + asyncResponse.resume(new RestException(e)); } } + // get matched namespaces; call unload for each namespaces; + private void filterAndUnloadMatchedNameSpaces(AsyncResponse asyncResponse, + NamespaceIsolationData policyData) throws Exception { + Namespaces namespaces = pulsar().getAdminClient().namespaces(); + + List nssToUnload = Lists.newArrayList(); + + pulsar().getAdminClient().tenants().getTenantsAsync() + .whenComplete((tenants, ex) -> { + if (ex != null) { + log.error("[{}] Failed to get tenants when setNamespaceIsolationPolicy.", clientAppId(), ex); + return; + } + AtomicInteger tenantsNumber = new AtomicInteger(tenants.size()); + // get all tenants now, for each tenants, get its namespaces + tenants.forEach(tenant -> namespaces.getNamespacesAsync(tenant) + .whenComplete((nss, e) -> { + int leftTenantsToHandle = tenantsNumber.decrementAndGet(); + if (ex != null) { + log.error("[{}] Failed to get namespaces for tenant {} when setNamespaceIsolationPolicy.", + clientAppId(), tenant, ex); + + if (leftTenantsToHandle == 0) { + unloadMatchedNamespacesList(asyncResponse, nssToUnload, namespaces); + } + + return; + } + + AtomicInteger nssNumber = new AtomicInteger(nss.size()); + + // get all namespaces for this tenant now. + nss.forEach(namespaceName -> { + int leftNssToHandle = nssNumber.decrementAndGet(); + + // if namespace match any policy regex, add it to ns list to be unload. + if (policyData.namespaces.stream() + .anyMatch(nsnameRegex -> namespaceName.matches(nsnameRegex))) { + nssToUnload.add(namespaceName); + } + + // all the tenants & namespaces get filtered. + if (leftNssToHandle == 0 && leftTenantsToHandle == 0) { + unloadMatchedNamespacesList(asyncResponse, nssToUnload, namespaces); + } + }); + })); + }); + } + + private void unloadMatchedNamespacesList(AsyncResponse asyncResponse, + List nssToUnload, + Namespaces namespaces) { + if (nssToUnload.size() == 0) { + asyncResponse.resume(Response.noContent().build()); + return; + } + + List> futures = nssToUnload.stream() + .map(namespaceName -> namespaces.unloadAsync(namespaceName)) + .collect(Collectors.toList()); + + FutureUtil.waitForAll(futures).whenComplete((result, exception) -> { + if (exception != null) { + log.error("[{}] Failed to unload namespace while setNamespaceIsolationPolicy.", + clientAppId(), exception); + asyncResponse.resume(new RestException(exception)); + return; + } + + try { + // write load info to load manager to make the load happens fast + pulsar().getLoadManager().get().writeLoadReportOnZookeeper(true); + } catch (Exception e) { + log.warn("[{}] Failed to writeLoadReportOnZookeeper.", clientAppId(), e); + } + + asyncResponse.resume(Response.noContent().build()); + return; + }); + } + private boolean createZnodeIfNotExist(String path, Optional value) throws KeeperException, InterruptedException { // create persistent node on ZooKeeper diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java index 11b3e2ec23112..7be18988222e7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java @@ -80,6 +80,14 @@ public interface LoadManager { */ void writeLoadReportOnZookeeper() throws Exception; + /** + * Publish the current load report on ZK, forced or not. + * By default rely on method writeLoadReportOnZookeeper(). + */ + default void writeLoadReportOnZookeeper(boolean force) throws Exception { + writeLoadReportOnZookeeper(); + } + /** * Update namespace bundle resource quota on ZK. */ diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java index 8a063a3727448..a778cdb688bcc 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java @@ -94,6 +94,13 @@ public interface ModularLoadManager { */ void writeBrokerDataOnZooKeeper(); + /** + * As any broker, write the local broker data to ZooKeeper, forced or not. + */ + default void writeBrokerDataOnZooKeeper(boolean force) { + writeBrokerDataOnZooKeeper(); + } + /** * As the leader broker, write bundle data aggregated from all brokers to ZooKeeper. */ diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java index c2e7cd16367f4..9df80e25cff2f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java @@ -992,16 +992,21 @@ private void updateLoadBalancingMetrics(final SystemResourceUsage systemResource */ @Override public void writeBrokerDataOnZooKeeper() { + writeBrokerDataOnZooKeeper(false); + } + + @Override + public void writeBrokerDataOnZooKeeper(boolean force) { try { updateLocalBrokerData(); - if (needBrokerDataUpdate()) { + if (needBrokerDataUpdate() || force) { localData.setLastUpdate(System.currentTimeMillis()); try { zkClient.setData(brokerZnodePath, localData.getJsonBytes(), -1); } catch (KeeperException.NoNodeException e) { ZkUtils.createFullPathOptimistic(zkClient, brokerZnodePath, localData.getJsonBytes(), - ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); + ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); } // Clear deltas. diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java index 48944ab6c50be..b8ee0737a41f9 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java @@ -114,6 +114,11 @@ public void writeLoadReportOnZookeeper() { loadManager.writeBrokerDataOnZooKeeper(); } + @Override + public void writeLoadReportOnZookeeper(boolean force) { + loadManager.writeBrokerDataOnZooKeeper(force); + } + @Override public void writeResourceQuotasToZooKeeper() { loadManager.writeBundleDataOnZooKeeper(); 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 index 5e3d9d6b05f62..7a6040611392b 100644 --- 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 @@ -100,6 +100,7 @@ public class AdminApiTest2 extends MockedPulsarServiceBaseTest { @Override public void setup() throws Exception { conf.setLoadBalancerEnabled(true); + conf.setEnableNamespaceIsolationUpdateOnTime(true); super.internalSetup(); // create otherbroker to test redirect on calls that need @@ -908,6 +909,62 @@ public void brokerNamespaceIsolationPolicies() throws Exception { assertFalse(isolationData.isPrimary); } + // create 1 namespace: + // 0. without isolation policy configured, lookup will success. + // 1. with matched isolation broker configured and matched, lookup will success. + // 2. update isolation policy, without broker matched, lookup will fail. + @Test + public void brokerNamespaceIsolationPoliciesUpdateOnTime() throws Exception { + String brokerName = pulsar.getAdvertisedAddress(); + String ns1Name = "prop-xyz/test_ns1_iso_" + System.currentTimeMillis(); + admin.namespaces().createNamespace(ns1Name, Sets.newHashSet("test")); + + // 0. without isolation policy configured, lookup will success. + String brokerUrl = admin.lookups().lookupTopic(ns1Name + "/topic1"); + assertTrue(brokerUrl.contains(brokerName)); + log.info("0 get lookup url {}", brokerUrl); + + // create + String policyName1 = "policy-1"; + String cluster = pulsar.getConfiguration().getClusterName(); + String namespaceRegex = ns1Name; + NamespaceIsolationData nsPolicyData1 = new NamespaceIsolationData(); + nsPolicyData1.namespaces = new ArrayList(); + nsPolicyData1.namespaces.add(ns1Name); + nsPolicyData1.primary = new ArrayList(); + nsPolicyData1.primary.add(brokerName + ".*"); + nsPolicyData1.auto_failover_policy = new AutoFailoverPolicyData(); + nsPolicyData1.auto_failover_policy.policy_type = AutoFailoverPolicyType.min_available; + nsPolicyData1.auto_failover_policy.parameters = new HashMap(); + nsPolicyData1.auto_failover_policy.parameters.put("min_limit", "1"); + nsPolicyData1.auto_failover_policy.parameters.put("usage_threshold", "100"); + admin.clusters().createNamespaceIsolationPolicyAsync(cluster, policyName1, nsPolicyData1).get(); + + // 1. with matched isolation broker configured and matched, lookup will success. + brokerUrl = admin.lookups().lookupTopic(ns1Name + "/topic2"); + assertTrue(brokerUrl.contains(brokerName)); + log.info(" 1 get lookup url {}", brokerUrl); + + // 2. update isolation policy, without broker matched, lookup will fail. + nsPolicyData1.primary = new ArrayList(); + nsPolicyData1.primary.add(brokerName + "not_match"); + admin.clusters().updateNamespaceIsolationPolicyAsync(cluster, policyName1, nsPolicyData1).get(); + + try { + admin.lookups().lookupTopic(ns1Name + "/topic3"); + } catch (Exception e) { + // expected lookup fail, because no brokers matched the policy. + log.info(" 2 expected fail lookup"); + } + + try { + admin.lookups().lookupTopic(ns1Name + "/topic1"); + } catch (Exception e) { + // expected lookup fail, because no brokers matched the policy. + log.info(" 22 expected fail lookup"); + } + } + @Test public void clustersList() throws PulsarAdminException { final String cluster = pulsar.getConfiguration().getClusterName(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java index d30b2e5c42983..c80e727dd532e 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java @@ -284,7 +284,8 @@ public void clusters() throws Exception { policyData.auto_failover_policy.parameters = new HashMap(); policyData.auto_failover_policy.parameters.put("min_limit", "1"); policyData.auto_failover_policy.parameters.put("usage_threshold", "90"); - clusters.setNamespaceIsolationPolicy("use", "policy1", policyData); + AsyncResponse response = mock(AsyncResponse.class); + clusters.setNamespaceIsolationPolicy(response,"use", "policy1", policyData); clusters.getNamespaceIsolationPolicies("use"); try {