Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[feat][admin] Add offload managedLedgerOffloadThreshold RestAPI and CLI tools #18218

Merged
merged 16 commits into from
Nov 21, 2022
Merged

[feat][admin] Add offload managedLedgerOffloadThreshold RestAPI and CLI tools #18218

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
286ad6c
review fix
dao-jun Oct 26, 2022
929f53b
fix
dao-jun Oct 27, 2022
ee633bb
Merge branch 'master' into dev/offload_trigger_admin
dao-jun Oct 27, 2022
c1b16b3
fix
dao-jun Oct 27, 2022
2f87e78
review fix
dao-jun Nov 1, 2022
2363e22
Review fix
dao-jun Nov 2, 2022
e31466b
Merge branch 'master' into dev/offload_trigger_admin
dao-jun Nov 2, 2022
9f9afd0
fix checkstyle
dao-jun Nov 15, 2022
504194b
Merge branch 'master' into dev/offload_trigger_admin
dao-jun Nov 15, 2022
fbd1f69
Merge branch 'master' into dev/offload_trigger_admin
dao-jun Nov 16, 2022
5520e2b
fix checkstyle
dao-jun Nov 16, 2022
b26a7b8
fix tests
dao-jun Nov 16, 2022
d172d84
fix tests
dao-jun Nov 17, 2022
547e36b
Merge branch 'master' into dev/offload_trigger_admin
dao-jun Nov 17, 2022
a22d3b3
fix checkstyle
dao-jun Nov 17, 2022
77b8032
fix tests
dao-jun Nov 18, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadLedgerDeleteTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,7 @@ Set<Long> deletedOffloads() {
OffloadPoliciesImpl.DEFAULT_MAX_BLOCK_SIZE_IN_BYTES,
OffloadPoliciesImpl.DEFAULT_READ_BUFFER_SIZE_IN_BYTES,
OffloadPoliciesImpl.DEFAULT_OFFLOAD_THRESHOLD_IN_BYTES,
OffloadPoliciesImpl.DEFAULT_OFFLOAD_THRESHOLD_IN_SECONDS,
OffloadPoliciesImpl.DEFAULT_OFFLOAD_DELETION_LAG_IN_MILLIS,
OffloadPoliciesImpl.DEFAULT_OFFLOADED_READ_PRIORITY);

Expand Down
3 changes: 2 additions & 1 deletion managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixReadTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -202,12 +202,13 @@ static class MockLedgerOffloader implements LedgerOffloader {
ConcurrentHashMap<UUID, ReadHandle> offloads = new ConcurrentHashMap<UUID, ReadHandle>();


OffloadPoliciesImpl offloadPolicies = OffloadPoliciesImpl.create("S3", "", "", "",
OffloadPoliciesImpl offloadPolicies = OffloadPoliciesImpl .create("S3", "", "", "",
null, null,
null, null,
OffloadPoliciesImpl.DEFAULT_MAX_BLOCK_SIZE_IN_BYTES,
OffloadPoliciesImpl.DEFAULT_READ_BUFFER_SIZE_IN_BYTES,
OffloadPoliciesImpl.DEFAULT_OFFLOAD_THRESHOLD_IN_BYTES,
OffloadPoliciesImpl.DEFAULT_OFFLOAD_THRESHOLD_IN_SECONDS,
OffloadPoliciesImpl.DEFAULT_OFFLOAD_DELETION_LAG_IN_MILLIS,
OffloadPoliciesImpl.DEFAULT_OFFLOADED_READ_PRIORITY);

Expand Down
1 change: 1 addition & 0 deletions managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1262,6 +1262,7 @@ Set<Long> deletedOffloads() {
OffloadPoliciesImpl.DEFAULT_MAX_BLOCK_SIZE_IN_BYTES,
OffloadPoliciesImpl.DEFAULT_READ_BUFFER_SIZE_IN_BYTES,
OffloadPoliciesImpl.DEFAULT_OFFLOAD_THRESHOLD_IN_BYTES,
OffloadPoliciesImpl.DEFAULT_OFFLOAD_THRESHOLD_IN_SECONDS,
OffloadPoliciesImpl.DEFAULT_OFFLOAD_DELETION_LAG_IN_MILLIS,
OffloadPoliciesImpl.DEFAULT_OFFLOADED_READ_PRIORITY);

Expand Down
26 changes: 26 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1965,6 +1965,32 @@ protected void internalSetOffloadThreshold(long newThreshold) {
}
}

protected void internalSetOffloadThresholdInSeconds(long newThreshold) {
tjiuming marked this conversation as resolved.
Show resolved Hide resolved
validateNamespacePolicyOperation(namespaceName, PolicyName.OFFLOAD, PolicyOperation.WRITE);
tjiuming marked this conversation as resolved.
Show resolved Hide resolved
validatePoliciesReadOnlyAccess();
tjiuming marked this conversation as resolved.
Show resolved Hide resolved

try {
updatePolicies(namespaceName, policies -> {
if (policies.offload_policies == null) {
policies.offload_policies = new OffloadPoliciesImpl();
}
((OffloadPoliciesImpl) policies.offload_policies)
.setManagedLedgerOffloadThresholdInSeconds(newThreshold);
policies.offload_threshold_in_seconds = newThreshold;
return policies;
});
log.info("[{}] Successfully updated offloadThresholdInSeconds configuration: namespace={}, value={}",
clientAppId(), namespaceName, newThreshold);

} catch (RestException pfe) {
throw pfe;
} catch (Exception e) {
log.error("[{}] Failed to update offloadThresholdInSeconds configuration for namespace {}",
clientAppId(), namespaceName, e);
throw new RestException(e);
}
}

protected void internalSetOffloadDeletionLag(Long newDeletionLagMs) {
validateNamespacePolicyOperation(namespaceName, PolicyName.OFFLOAD, PolicyOperation.WRITE);
validatePoliciesReadOnlyAccess();
Expand Down
44 changes: 44 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2088,6 +2088,50 @@ public void setOffloadThreshold(@PathParam("tenant") String tenant,
internalSetOffloadThreshold(newThreshold);
}

@GET
@Path("/{tenant}/{namespace}/offloadThresholdInSeconds")
@ApiOperation(value = "Maximum number of bytes stored on the pulsar cluster for a topic,"
+ " before the broker will start offloading to longterm storage",
notes = "A negative value disables automatic offloading")
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Namespace doesn't exist") })
public void getOffloadThresholdInSeconds(
@Suspended final AsyncResponse asyncResponse,
@PathParam("tenant") String tenant,
@PathParam("namespace") String namespace) {
validateNamespaceName(tenant, namespace);
validateNamespacePolicyOperationAsync(namespaceName, PolicyName.OFFLOAD, PolicyOperation.READ)
.thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
.thenAccept(policies -> {
if (policies.offload_policies == null) {
asyncResponse.resume(policies.offload_threshold_in_seconds);
} else {
asyncResponse.resume(policies.offload_policies.getManagedLedgerOffloadThresholdInSeconds());
}
})
.exceptionally(ex -> {
log.error("[{}] Failed to get offload threshold on namespace {}", clientAppId(), namespaceName, ex);
resumeAsyncResponseExceptionally(asyncResponse, ex);
return null;
});
}

@PUT
@Path("/{tenant}/{namespace}/offloadThresholdInSeconds")
@ApiOperation(value = "Set maximum number of bytes stored on the pulsar cluster for a topic,"
+ " before the broker will start offloading to longterm storage",
notes = "A negative value disables automatic offloading")
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Namespace doesn't exist"),
@ApiResponse(code = 409, message = "Concurrent modification"),
@ApiResponse(code = 412, message = "offloadThreshold value is not valid") })
public void setOffloadThresholdInSeconds(@PathParam("tenant") String tenant,
@PathParam("namespace") String namespace,
long newThreshold) {
validateNamespaceName(tenant, namespace);
internalSetOffloadThresholdInSeconds(newThreshold);
}

@GET
@Path("/{tenant}/{namespace}/offloadDeletionLagMs")
@ApiOperation(value = "Number of milliseconds to wait before deleting a ledger segment which has been offloaded"
Expand Down
101 changes: 100 additions & 1 deletion pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiOffloadTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,9 +52,14 @@
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

import org.apache.bookkeeper.mledger.LedgerOffloader;
import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.admin.LongRunningProcessStatus;
import org.apache.pulsar.client.admin.PulsarAdminException.ConflictException;
Expand Down Expand Up @@ -186,12 +191,13 @@ public void testOffloadPolicies() throws Exception {
String bucket = "test-bucket";
String endpoint = "test-endpoint";
long offloadThresholdInBytes = 0;
long offloadThresholdInSeconds = 100;
long offloadDeletionLagInMillis = 100L;
OffloadedReadPriority priority = OffloadedReadPriority.TIERED_STORAGE_FIRST;

OffloadPolicies offload1 = OffloadPoliciesImpl.create(
driver, region, bucket, endpoint, null, null, null, null,
100, 100, offloadThresholdInBytes, offloadDeletionLagInMillis, priority);
100, 100, offloadThresholdInBytes, offloadThresholdInSeconds, offloadDeletionLagInMillis, priority);
admin.namespaces().setOffloadPolicies(namespaceName, offload1);
OffloadPolicies offload2 = admin.namespaces().getOffloadPolicies(namespaceName);
assertEquals(offload1, offload2);
Expand Down Expand Up @@ -239,6 +245,7 @@ public void testOffloadPoliciesAppliedApi() throws Exception {

OffloadPoliciesImpl namespacePolicies = new OffloadPoliciesImpl();
namespacePolicies.setManagedLedgerOffloadThresholdInBytes(100L);
namespacePolicies.setManagedLedgerOffloadThresholdInSeconds(100L);
namespacePolicies.setManagedLedgerOffloadDeletionLagInMillis(200L);
namespacePolicies.setManagedLedgerOffloadDriver("s3");
namespacePolicies.setManagedLedgerOffloadBucket("buck");
Expand All @@ -250,6 +257,7 @@ public void testOffloadPoliciesAppliedApi() throws Exception {

OffloadPoliciesImpl topicPolicies = new OffloadPoliciesImpl();
topicPolicies.setManagedLedgerOffloadThresholdInBytes(200L);
topicPolicies.setManagedLedgerOffloadThresholdInSeconds(200L);
topicPolicies.setManagedLedgerOffloadDeletionLagInMillis(400L);
topicPolicies.setManagedLedgerOffloadDriver("s3");
topicPolicies.setManagedLedgerOffloadBucket("buck2");
Expand All @@ -267,6 +275,97 @@ public void testOffloadPoliciesAppliedApi() throws Exception {
-> assertEquals(admin.topics().getOffloadPolicies(topicName, true), brokerPolicies));
}


@Test
public void testSetNamespaceOffloadPolicies() throws Exception {
conf.setManagedLedgerOffloadThresholdInSeconds(100);
conf.setManagedLedgerOffloadAutoTriggerSizeThresholdBytes(100);

OffloadPoliciesImpl policies = new OffloadPoliciesImpl();
policies.setManagedLedgerOffloadThresholdInBytes(200L);
policies.setManagedLedgerOffloadThresholdInSeconds(200L);
policies.setManagedLedgerOffloadDeletionLagInMillis(400L);
policies.setManagedLedgerOffloadDriver("s3");
policies.setManagedLedgerOffloadBucket("buck2");

admin.namespaces().setOffloadThresholdInSeconds(myNamespace, 300);
assertEquals(300, admin.namespaces().getOffloadThresholdInSeconds(myNamespace));
tjiuming marked this conversation as resolved.
Show resolved Hide resolved

admin.namespaces().setOffloadPolicies(myNamespace, policies);
assertEquals(policies, admin.namespaces().getOffloadPolicies(myNamespace));
tjiuming marked this conversation as resolved.
Show resolved Hide resolved

String topicName = testTopic + UUID.randomUUID();
try {
Topic topic = pulsar.getBrokerService().getOrCreateTopic(topicName).get(10, TimeUnit.SECONDS);
assertNotNull(topic);

assertTrue(topic instanceof PersistentTopic);

PersistentTopic persistentTopic = (PersistentTopic) topic;
ManagedLedger ledger = persistentTopic.getManagedLedger();
ManagedLedgerConfig config = ledger.getConfig();
OffloadPolicies policies1 = config.getLedgerOffloader().getOffloadPolicies();

assertEquals(policies1.getManagedLedgerOffloadThresholdInBytes(), policies.getManagedLedgerOffloadThresholdInBytes());
assertEquals(policies1.getManagedLedgerOffloadThresholdInSeconds(), policies.getManagedLedgerOffloadThresholdInSeconds());
} finally {
pulsar.getBrokerService().deleteTopic(topicName, true);
}
}

@Test
public void testSetTopicOffloadPolicies() throws Exception {
conf.setManagedLedgerOffloadThresholdInSeconds(100);
conf.setManagedLedgerOffloadAutoTriggerSizeThresholdBytes(100);

OffloadPoliciesImpl namespacePolicies = new OffloadPoliciesImpl();
namespacePolicies.setManagedLedgerOffloadThresholdInBytes(200L);
namespacePolicies.setManagedLedgerOffloadThresholdInSeconds(200L);
namespacePolicies.setManagedLedgerOffloadDeletionLagInMillis(400L);
namespacePolicies.setManagedLedgerOffloadDriver("s3");
namespacePolicies.setManagedLedgerOffloadBucket("buck2");

admin.namespaces().setOffloadThresholdInSeconds(myNamespace, 300);
assertEquals(300, admin.namespaces().getOffloadThresholdInSeconds(myNamespace));
tjiuming marked this conversation as resolved.
Show resolved Hide resolved

admin.namespaces().setOffloadPolicies(myNamespace, namespacePolicies);
assertEquals(namespacePolicies, admin.namespaces().getOffloadPolicies(myNamespace));
tjiuming marked this conversation as resolved.
Show resolved Hide resolved

OffloadPoliciesImpl topicPolicies = new OffloadPoliciesImpl();
topicPolicies.setManagedLedgerOffloadThresholdInBytes(500L);
topicPolicies.setManagedLedgerOffloadThresholdInSeconds(500L);
topicPolicies.setManagedLedgerOffloadDeletionLagInMillis(400L);
topicPolicies.setManagedLedgerOffloadDriver("s3");
topicPolicies.setManagedLedgerOffloadBucket("buck2");

String topicName = testTopic + UUID.randomUUID();
admin.topicPolicies().setOffloadPolicies(topicName, topicPolicies);

assertEquals(admin.topicPolicies().getOffloadPolicies(topicName).getManagedLedgerOffloadThresholdInSeconds(),
topicPolicies.getManagedLedgerOffloadThresholdInSeconds());
assertEquals(admin.topicPolicies().getOffloadPolicies(topicName).getManagedLedgerOffloadThresholdInBytes(),
topicPolicies.getManagedLedgerOffloadThresholdInBytes());

try {
Topic topic = pulsar.getBrokerService().getOrCreateTopic(topicName).get(10, TimeUnit.SECONDS);
assertNotNull(topic);

assertTrue(topic instanceof PersistentTopic);

PersistentTopic persistentTopic = (PersistentTopic) topic;
ManagedLedger ledger = persistentTopic.getManagedLedger();
ManagedLedgerConfig config = ledger.getConfig();
OffloadPolicies policies1 = config.getLedgerOffloader().getOffloadPolicies();

assertEquals(policies1.getManagedLedgerOffloadThresholdInBytes(),
topicPolicies.getManagedLedgerOffloadThresholdInBytes());
assertEquals(policies1.getManagedLedgerOffloadThresholdInSeconds(),
topicPolicies.getManagedLedgerOffloadThresholdInSeconds());
} finally {
pulsar.getBrokerService().deleteTopic(topicName, true);
}
}

@Test
public void testTopicLevelOffloadPartitioned() throws Exception {
testOffload(true);
Expand Down
Loading