Visitar URL original
[improve][broker] PIP-192 Excluded bundles with isolation policy or a… · apache/pulsar@17f6bcd · GitHub
Skip to content

Commit 17f6bcd

Browse files
author
Heesung Sohn
committed
[improve][broker] PIP-192 Excluded bundles with isolation policy or anti-affinity-group policy from topk load bundles
1 parent da3cab5 commit 17f6bcd

7 files changed

Lines changed: 312 additions & 54 deletions

File tree

‎pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2600,6 +2600,16 @@ The delayed message index time step(in seconds) in per bucket snapshot segment,
26002600
)
26012601
private long loadBalancerServiceUnitStateCleanUpDelayTimeInSeconds = 604800;
26022602

2603+
@FieldContext(
2604+
category = CATEGORY_LOAD_BALANCER,
2605+
dynamic = true,
2606+
doc = "Option to automatically unload namespace bundles with affinity(isolation) "
2607+
+ "or anti-affinity group policies."
2608+
+ "Such bundles are not ideal targets to auto-unload as destination brokers are limited."
2609+
+ "(only used in load balancer extension logics)"
2610+
)
2611+
private boolean loadBalancerSheddingBundlesWithPoliciesEnabled = false;
2612+
26032613
/**** --- Replication. --- ****/
26042614
@FieldContext(
26052615
category = CATEGORY_REPLICATION,

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/models/TopKBundles.java‎

Lines changed: 60 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,14 @@
2424
import java.util.Map;
2525
import lombok.EqualsAndHashCode;
2626
import lombok.Getter;
27-
import lombok.NoArgsConstructor;
2827
import lombok.ToString;
28+
import lombok.extern.slf4j.Slf4j;
29+
import org.apache.pulsar.broker.PulsarService;
2930
import org.apache.pulsar.broker.loadbalance.extensions.data.TopBundlesLoadData;
31+
import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared;
32+
import org.apache.pulsar.broker.loadbalance.impl.SimpleResourceAllocationPolicies;
3033
import org.apache.pulsar.common.naming.NamespaceName;
34+
import org.apache.pulsar.metadata.api.MetadataStoreException;
3135
import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
3236

3337
/**
@@ -36,14 +40,23 @@
3640
@Getter
3741
@ToString
3842
@EqualsAndHashCode
39-
@NoArgsConstructor
43+
@Slf4j
4044
public class TopKBundles {
4145

4246
// temp array for sorting
4347
private final List<Map.Entry<String, ? extends Comparable>> arr = new ArrayList<>();
4448

4549
private final TopBundlesLoadData loadData = new TopBundlesLoadData();
4650

51+
private final PulsarService pulsar;
52+
53+
private final SimpleResourceAllocationPolicies allocationPolicies;
54+
55+
public TopKBundles(PulsarService pulsar) {
56+
this.pulsar = pulsar;
57+
this.allocationPolicies = new SimpleResourceAllocationPolicies(pulsar);
58+
}
59+
4760
/**
4861
* Update the topK bundles from the input bundleStats.
4962
*
@@ -52,26 +65,35 @@ public class TopKBundles {
5265
*/
5366
public void update(Map<String, NamespaceBundleStats> bundleStats, int topk) {
5467
arr.clear();
55-
for (var etr : bundleStats.entrySet()) {
56-
if (etr.getKey().startsWith(NamespaceName.SYSTEM_NAMESPACE.toString())) {
57-
continue;
68+
try {
69+
var isLoadBalancerSheddingBundlesWithPoliciesEnabled =
70+
pulsar.getConfiguration().isLoadBalancerSheddingBundlesWithPoliciesEnabled();
71+
for (var etr : bundleStats.entrySet()) {
72+
String bundle = etr.getKey();
73+
if (bundle.startsWith(NamespaceName.SYSTEM_NAMESPACE.toString())) {
74+
continue;
75+
}
76+
if (!isLoadBalancerSheddingBundlesWithPoliciesEnabled && hasPolicies(bundle)) {
77+
continue;
78+
}
79+
arr.add(etr);
5880
}
59-
arr.add(etr);
60-
}
61-
var topKBundlesLoadData = loadData.getTopBundlesLoadData();
62-
topKBundlesLoadData.clear();
63-
if (arr.isEmpty()) {
64-
return;
65-
}
66-
topk = Math.min(topk, arr.size());
67-
partitionSort(arr, topk);
81+
var topKBundlesLoadData = loadData.getTopBundlesLoadData();
82+
topKBundlesLoadData.clear();
83+
if (arr.isEmpty()) {
84+
return;
85+
}
86+
topk = Math.min(topk, arr.size());
87+
partitionSort(arr, topk);
6888

69-
for (int i = 0; i < topk; i++) {
70-
var etr = arr.get(i);
71-
topKBundlesLoadData.add(
72-
new TopBundlesLoadData.BundleLoadData(etr.getKey(), (NamespaceBundleStats) etr.getValue()));
89+
for (int i = 0; i < topk; i++) {
90+
var etr = arr.get(i);
91+
topKBundlesLoadData.add(
92+
new TopBundlesLoadData.BundleLoadData(etr.getKey(), (NamespaceBundleStats) etr.getValue()));
93+
}
94+
} finally {
95+
arr.clear();
7396
}
74-
arr.clear();
7597
}
7698

7799
static void partitionSort(List<Map.Entry<String, ? extends Comparable>> arr, int k) {
@@ -109,4 +131,23 @@ static void partitionSort(List<Map.Entry<String, ? extends Comparable>> arr, int
109131
}
110132
Collections.sort(arr.subList(0, end), (a, b) -> b.getValue().compareTo(a.getValue()));
111133
}
134+
135+
private boolean hasPolicies(String bundle) {
136+
NamespaceName namespace = NamespaceName.get(LoadManagerShared.getNamespaceNameFromBundleName(bundle));
137+
if (allocationPolicies.areIsolationPoliciesPresent(namespace)) {
138+
return true;
139+
}
140+
141+
try {
142+
var antiAffinityGroupOptional =
143+
LoadManagerShared.getNamespaceAntiAffinityGroup(pulsar, namespace.toString());
144+
if (antiAffinityGroupOptional.isPresent()) {
145+
return true;
146+
}
147+
} catch (MetadataStoreException e) {
148+
log.error("Failed to get localPolicies for bundle:{}.", bundle, e);
149+
throw new RuntimeException(e);
150+
}
151+
return false;
152+
}
112153
}

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/TopBundleLoadDataReporter.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ public TopBundleLoadDataReporter(PulsarService pulsar,
4848
this.lookupServiceAddress = lookupServiceAddress;
4949
this.bundleLoadDataStore = bundleLoadDataStore;
5050
this.lastBundleStatsUpdatedAt = 0;
51-
this.topKBundles = new TopKBundles();
51+
this.topKBundles = new TopKBundles(pulsar);
5252
}
5353

5454
@Override

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedder.java‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -444,17 +444,21 @@ private boolean isTransferable(LoadManagerContext context,
444444
if (pulsar == null || allocationPolicies == null) {
445445
return true;
446446
}
447+
var isLoadBalancerSheddingBundlesWithPoliciesEnabled =
448+
context.brokerConfiguration().isLoadBalancerSheddingBundlesWithPoliciesEnabled();
447449
String namespace = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
448450
final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
449451
NamespaceBundle namespaceBundle =
450452
pulsar.getNamespaceService().getNamespaceBundleFactory().getBundle(namespace, bundleRange);
451453

452-
if (!canTransferWithIsolationPoliciesToBroker(
454+
if (!isLoadBalancerSheddingBundlesWithPoliciesEnabled
455+
&& !canTransferWithIsolationPoliciesToBroker(
453456
context, availableBrokers, namespaceBundle, srcBroker, dstBroker)) {
454457
return false;
455458
}
456459

457-
if (!antiAffinityGroupPolicyHelper.canUnload(availableBrokers, bundle, srcBroker, dstBroker)) {
460+
if (!isLoadBalancerSheddingBundlesWithPoliciesEnabled
461+
&& !antiAffinityGroupPolicyHelper.canUnload(availableBrokers, bundle, srcBroker, dstBroker)) {
458462
return false;
459463
}
460464

‎pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/models/TopKBundlesTest.java‎

Lines changed: 163 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -18,65 +18,215 @@
1818
*/
1919
package org.apache.pulsar.broker.loadbalance.extensions.models;
2020

21+
import static org.mockito.ArgumentMatchers.eq;
22+
import static org.mockito.ArgumentMatchers.any;
23+
import static org.mockito.Mockito.doReturn;
24+
import static org.mockito.Mockito.mock;
2125
import static org.testng.Assert.assertEquals;
2226
import static org.testng.Assert.assertNull;
23-
import static org.testng.Assert.assertTrue;
2427
import java.util.ArrayList;
2528
import java.util.Collections;
2629
import java.util.HashMap;
2730
import java.util.List;
2831
import java.util.Map;
32+
import java.util.Optional;
2933
import java.util.Random;
34+
import org.apache.pulsar.broker.PulsarService;
35+
import org.apache.pulsar.broker.ServiceConfiguration;
36+
import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared;
37+
import org.apache.pulsar.broker.resources.LocalPoliciesResources;
38+
import org.apache.pulsar.broker.resources.NamespaceResources;
39+
import org.apache.pulsar.broker.resources.PulsarResources;
40+
import org.apache.pulsar.common.naming.NamespaceName;
41+
import org.apache.pulsar.common.policies.data.AutoFailoverPolicyData;
42+
import org.apache.pulsar.common.policies.data.AutoFailoverPolicyType;
43+
import org.apache.pulsar.common.policies.data.LocalPolicies;
44+
import org.apache.pulsar.common.policies.data.NamespaceIsolationData;
45+
import org.apache.pulsar.common.policies.data.NamespaceIsolationDataImpl;
46+
import org.apache.pulsar.common.policies.impl.NamespaceIsolationPolicies;
47+
import org.apache.pulsar.metadata.api.MetadataStoreException;
3048
import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
49+
import org.testng.annotations.BeforeMethod;
3150
import org.testng.annotations.Test;
3251

3352
@Test(groups = "broker")
3453
public class TopKBundlesTest {
54+
private PulsarService pulsar;
55+
private ServiceConfiguration configuration;
56+
private NamespaceResources.IsolationPolicyResources isolationPolicyResources;
57+
private PulsarResources pulsarResources;
58+
private LocalPoliciesResources localPoliciesResources;
3559

60+
String bundle1 = "my-tenant/my-namespace1/0x00000000_0x0FFFFFFF";
61+
String bundle2 = "my-tenant/my-namespace2/0x00000000_0x0FFFFFFF";
62+
String bundle3 = "my-tenant/my-namespace3/0x00000000_0x0FFFFFFF";
63+
String bundle4 = "my-tenant/my-namespace4/0x00000000_0x0FFFFFFF";
64+
65+
@BeforeMethod
66+
public void init() throws MetadataStoreException {
67+
pulsar = mock(PulsarService.class);
68+
configuration = new ServiceConfiguration();
69+
doReturn(configuration).when(pulsar).getConfiguration();
70+
configuration.setLoadBalancerSheddingBundlesWithPoliciesEnabled(false);
71+
pulsarResources = mock(PulsarResources.class);
72+
var namespaceResources = mock(NamespaceResources.class);
73+
74+
isolationPolicyResources = mock(NamespaceResources.IsolationPolicyResources.class);
75+
doReturn(pulsarResources).when(pulsar).getPulsarResources();
76+
doReturn(namespaceResources).when(pulsarResources).getNamespaceResources();
77+
doReturn(isolationPolicyResources).when(namespaceResources).getIsolationPolicies();
78+
doReturn(Optional.empty()).when(isolationPolicyResources).getIsolationDataPolicies(any());
79+
80+
localPoliciesResources = mock(LocalPoliciesResources.class);
81+
doReturn(localPoliciesResources).when(pulsarResources).getLocalPolicies();
82+
doReturn(Optional.empty()).when(localPoliciesResources).getLocalPolicies(any());
83+
84+
}
3685
@Test
3786
public void testTopBundlesLoadData() {
3887
Map<String, NamespaceBundleStats> bundleStats = new HashMap<>();
39-
var topKBundles = new TopKBundles();
88+
var topKBundles = new TopKBundles(pulsar);
4089
NamespaceBundleStats stats1 = new NamespaceBundleStats();
4190
stats1.msgRateIn = 500;
42-
bundleStats.put("bundle-1", stats1);
91+
bundleStats.put(bundle1, stats1);
4392

4493
NamespaceBundleStats stats2 = new NamespaceBundleStats();
4594
stats2.msgRateIn = 10000;
46-
bundleStats.put("bundle-2", stats2);
95+
bundleStats.put(bundle2, stats2);
4796

4897
NamespaceBundleStats stats3 = new NamespaceBundleStats();
4998
stats3.msgRateIn = 100000;
50-
bundleStats.put("bundle-3", stats3);
99+
bundleStats.put(bundle3, stats3);
51100

52101
NamespaceBundleStats stats4 = new NamespaceBundleStats();
53102
stats4.msgRateIn = 0;
54-
bundleStats.put("bundle-4", stats4);
103+
bundleStats.put(bundle4, stats4);
55104

56105
topKBundles.update(bundleStats, 3);
57106
var top0 = topKBundles.getLoadData().getTopBundlesLoadData().get(0);
58107
var top1 = topKBundles.getLoadData().getTopBundlesLoadData().get(1);
59108
var top2 = topKBundles.getLoadData().getTopBundlesLoadData().get(2);
60109

61-
assertEquals(top0.bundleName(), "bundle-3");
62-
assertEquals(top1.bundleName(), "bundle-2");
63-
assertEquals(top2.bundleName(), "bundle-1");
110+
assertEquals(top0.bundleName(), bundle3);
111+
assertEquals(top1.bundleName(), bundle2);
112+
assertEquals(top2.bundleName(), bundle1);
64113
}
65114

66115
@Test
67116
public void testSystemNamespace() {
68117
Map<String, NamespaceBundleStats> bundleStats = new HashMap<>();
69-
var topKBundles = new TopKBundles();
118+
var topKBundles = new TopKBundles(pulsar);
70119
NamespaceBundleStats stats1 = new NamespaceBundleStats();
71120
stats1.msgRateIn = 500;
72-
bundleStats.put("pulsar/system/bundle-1", stats1);
121+
bundleStats.put("pulsar/system/0x00000000_0x0FFFFFFF", stats1);
73122

74123
NamespaceBundleStats stats2 = new NamespaceBundleStats();
75124
stats2.msgRateIn = 10000;
76-
bundleStats.put("pulsar/system/bundle-2", stats2);
125+
bundleStats.put(bundle1, stats2);
77126

78127
topKBundles.update(bundleStats, 2);
79-
assertTrue(topKBundles.getLoadData().getTopBundlesLoadData().isEmpty());
128+
129+
assertEquals(topKBundles.getLoadData().getTopBundlesLoadData().size(), 1);
130+
var top0 = topKBundles.getLoadData().getTopBundlesLoadData().get(0);
131+
assertEquals(top0.bundleName(), bundle1);
132+
}
133+
134+
135+
private void setAntiAffinityGroup() throws MetadataStoreException {
136+
LocalPolicies localPolicies = new LocalPolicies(null, null, "namespaceAntiAffinityGroup");
137+
NamespaceName namespace = NamespaceName.get(LoadManagerShared.getNamespaceNameFromBundleName(bundle2));
138+
doReturn(Optional.of(localPolicies)).when(localPoliciesResources).getLocalPolicies(eq(namespace));
139+
}
140+
141+
private void setIsolationPolicy() throws MetadataStoreException {
142+
Map<String, String> parameters = new HashMap<>();
143+
parameters.put("min_limit", "3");
144+
parameters.put("usage_threshold", "90");
145+
var policyData = Map.of("policy", (NamespaceIsolationDataImpl)
146+
NamespaceIsolationData.builder()
147+
.namespaces(Collections.singletonList("my-tenant/my-namespace1.*"))
148+
.primary(Collections.singletonList("prod1-broker[1-3].messaging.use.example.com"))
149+
.secondary(Collections.singletonList("prod1-broker.*.use.example.com"))
150+
.autoFailoverPolicy(AutoFailoverPolicyData.builder()
151+
.policyType(AutoFailoverPolicyType.min_available)
152+
.parameters(parameters)
153+
.build()
154+
).build());
155+
156+
NamespaceIsolationPolicies policies = new NamespaceIsolationPolicies(policyData);
157+
doReturn(Optional.of(policies)).when(isolationPolicyResources).getIsolationDataPolicies(any());
158+
}
159+
160+
@Test
161+
public void testIsolationPolicy() throws MetadataStoreException {
162+
163+
setIsolationPolicy();
164+
165+
Map<String, NamespaceBundleStats> bundleStats = new HashMap<>();
166+
var topKBundles = new TopKBundles(pulsar);
167+
NamespaceBundleStats stats1 = new NamespaceBundleStats();
168+
stats1.msgRateIn = 500;
169+
bundleStats.put(bundle1, stats1);
170+
171+
NamespaceBundleStats stats2 = new NamespaceBundleStats();
172+
stats2.msgRateIn = 10000;
173+
bundleStats.put(bundle2, stats2);
174+
175+
topKBundles.update(bundleStats, 2);
176+
assertEquals(topKBundles.getLoadData().getTopBundlesLoadData().size(), 1);
177+
var top0 = topKBundles.getLoadData().getTopBundlesLoadData().get(0);
178+
assertEquals(top0.bundleName(), bundle2);
179+
}
180+
181+
182+
@Test
183+
public void testAntiAffinityGroupPolicy() throws MetadataStoreException {
184+
185+
setAntiAffinityGroup();
186+
187+
Map<String, NamespaceBundleStats> bundleStats = new HashMap<>();
188+
var topKBundles = new TopKBundles(pulsar);
189+
NamespaceBundleStats stats1 = new NamespaceBundleStats();
190+
stats1.msgRateIn = 500;
191+
bundleStats.put(bundle1, stats1);
192+
193+
NamespaceBundleStats stats2 = new NamespaceBundleStats();
194+
stats2.msgRateIn = 10000;
195+
bundleStats.put(bundle2, stats2);
196+
197+
topKBundles.update(bundleStats, 2);
198+
assertEquals(topKBundles.getLoadData().getTopBundlesLoadData().size(), 1);
199+
var top0 = topKBundles.getLoadData().getTopBundlesLoadData().get(0);
200+
assertEquals(top0.bundleName(), bundle1);
201+
202+
}
203+
204+
@Test
205+
public void testLoadBalancerSheddingBundlesWithPoliciesEnabledConfig() throws MetadataStoreException {
206+
207+
setIsolationPolicy();
208+
setAntiAffinityGroup();
209+
210+
configuration.setLoadBalancerSheddingBundlesWithPoliciesEnabled(true);
211+
212+
Map<String, NamespaceBundleStats> bundleStats = new HashMap<>();
213+
var topKBundles = new TopKBundles(pulsar);
214+
NamespaceBundleStats stats1 = new NamespaceBundleStats();
215+
stats1.msgRateIn = 500;
216+
bundleStats.put(bundle1, stats1);
217+
218+
NamespaceBundleStats stats2 = new NamespaceBundleStats();
219+
stats2.msgRateIn = 10000;
220+
bundleStats.put(bundle2, stats2);
221+
222+
topKBundles.update(bundleStats, 2);
223+
224+
assertEquals(topKBundles.getLoadData().getTopBundlesLoadData().size(), 2);
225+
var top0 = topKBundles.getLoadData().getTopBundlesLoadData().get(0);
226+
var top1 = topKBundles.getLoadData().getTopBundlesLoadData().get(1);
227+
228+
assertEquals(top0.bundleName(), bundle2);
229+
assertEquals(top1.bundleName(), bundle1);
80230
}
81231

82232

0 commit comments

Comments
 (0)