You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bookkeeper.apache.org by iv...@apache.org on 2020/03/04 10:36:58 UTC
[bookkeeper] branch recopy-ledger created (now c89166c)
This is an automated email from the ASF dual-hosted git repository.
ivank pushed a change to branch recopy-ledger
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git.
at c89166c Tools for migrating ledgers
This branch includes the following new commits:
new c89166c Tools for migrating ledgers
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[bookkeeper] 01/01: Tools for migrating ledgers
Posted by iv...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ivank pushed a commit to branch recopy-ledger
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
commit c89166ca0441ae8d7a6d4cc3d4aff6c200fc5469
Author: Ivan Kelly <iv...@apache.org>
AuthorDate: Wed Mar 4 11:32:59 2020 +0100
Tools for migrating ledgers
Tools for migrating ledgers from one set of bookies to another.
- GenerateChecksums: Generates a checksum of the ledger metadata and
ledger entries, and stores to a file
- VerifyChecksums: Verify that the ledger on the bookies matches
previously generated checksums.
- RecopyLedgers: Copy ledger contents from one bookie to another, and
update the metadata.
- ReferencesBookies: List all bookies references by ledgers.
---
.../org/apache/bookkeeper/client/BookKeeper.java | 2 +-
.../apache/bookkeeper/client/BookieWatcher.java | 2 +-
.../apache/bookkeeper/client/ClientContext.java | 2 +-
.../bookkeeper/client/ClientInternalConf.java | 2 +-
.../org/apache/bookkeeper/client/LedgerHandle.java | 6 +-
.../apache/bookkeeper/client/LedgerHandleAdv.java | 2 +-
.../bookkeeper/client/LedgerMetadataBuilder.java | 10 ++
.../cli/commands/client/GenerateChecksums.java | 135 +++++++++++++++
.../tools/cli/commands/client/LedgerRewriter.java | 73 ++++++++
.../cli/commands/client/RecopyLedgerCommand.java | 192 +++++++++++++++++++++
.../cli/commands/client/ReferencedBookies.java | 115 ++++++++++++
.../tools/cli/commands/client/VerifyChecksums.java | 117 +++++++++++++
.../bookkeeper/tests/integration/TestCLI.java | 149 ++++++++++------
.../tools/cli/commands/LedgerCommandGroup.java | 8 +
14 files changed, 751 insertions(+), 64 deletions(-)
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
index 94591af..96b5f7a 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
@@ -1505,7 +1505,7 @@ public class BookKeeper implements org.apache.bookkeeper.client.api.BookKeeper {
}
};
- ClientContext getClientCtx() {
+ public ClientContext getClientCtx() {
return clientCtx;
}
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookieWatcher.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookieWatcher.java
index b881569..da7008a 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookieWatcher.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookieWatcher.java
@@ -24,7 +24,7 @@ import java.util.Set;
import org.apache.bookkeeper.client.BKException.BKNotEnoughBookiesException;
import org.apache.bookkeeper.net.BookieSocketAddress;
-interface BookieWatcher {
+public interface BookieWatcher {
Set<BookieSocketAddress> getBookies() throws BKException;
Set<BookieSocketAddress> getAllBookies() throws BKException;
Set<BookieSocketAddress> getReadOnlyBookies() throws BKException;
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ClientContext.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ClientContext.java
index da3abde..aef3e08 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ClientContext.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ClientContext.java
@@ -33,7 +33,7 @@ import org.apache.bookkeeper.proto.BookieClient;
* but they are present to the LedgerHandle through this interface to allow
* tests to easily inject mocked versions.
*/
-interface ClientContext {
+public interface ClientContext {
ClientInternalConf getConf();
LedgerManager getLedgerManager();
BookieWatcher getBookieWatcher();
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ClientInternalConf.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ClientInternalConf.java
index 85d42c6..c90824d 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ClientInternalConf.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ClientInternalConf.java
@@ -29,7 +29,7 @@ import org.apache.bookkeeper.feature.Feature;
import org.apache.bookkeeper.feature.FeatureProvider;
import org.apache.bookkeeper.feature.SettableFeatureProvider;
-class ClientInternalConf {
+public class ClientInternalConf {
final Feature disableEnsembleChangeFeature;
final boolean delayEnsembleChange;
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
index f3abef8..4dc168e 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
@@ -357,7 +357,7 @@ public class LedgerHandle implements WriteHandle {
return versionedMetadata.getValue();
}
- Versioned<LedgerMetadata> getVersionedLedgerMetadata() {
+ public Versioned<LedgerMetadata> getVersionedLedgerMetadata() {
return versionedMetadata;
}
@@ -416,7 +416,7 @@ public class LedgerHandle implements WriteHandle {
*
* @return DigestManager for the LedgerHandle
*/
- DigestManager getDigestManager() {
+ public DigestManager getDigestManager() {
return macManager;
}
@@ -455,7 +455,7 @@ public class LedgerHandle implements WriteHandle {
*
* @return DistributionSchedule for the LedgerHandle
*/
- DistributionSchedule getDistributionSchedule() {
+ public DistributionSchedule getDistributionSchedule() {
return distributionSchedule;
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
index 14317d0..20161ed 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
@@ -57,7 +57,7 @@ public class LedgerHandleAdv extends LedgerHandle implements WriteAdvHandle {
}
}
- LedgerHandleAdv(ClientContext clientCtx,
+ public LedgerHandleAdv(ClientContext clientCtx,
long ledgerId, Versioned<LedgerMetadata> metadata,
BookKeeper.DigestType digestType, byte[] password, EnumSet<WriteFlag> writeFlags)
throws GeneralSecurityException, NumberFormatException {
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerMetadataBuilder.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerMetadataBuilder.java
index e4f75ce..df58ba4 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerMetadataBuilder.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerMetadataBuilder.java
@@ -149,6 +149,16 @@ public class LedgerMetadataBuilder {
return this;
}
+ public LedgerMetadataBuilder clearEnsembles() {
+ ensembles.clear();
+ return this;
+ }
+
+ public LedgerMetadataBuilder withOpenState() {
+ this.state = State.OPEN;
+ return this;
+ }
+
public LedgerMetadataBuilder withInRecoveryState() {
this.state = State.IN_RECOVERY;
return this;
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/client/GenerateChecksums.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/client/GenerateChecksums.java
new file mode 100644
index 0000000..24e870d
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/client/GenerateChecksums.java
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.tools.cli.commands.client;
+import org.apache.bookkeeper.client.api.WriteFlag;
+
+import com.beust.jcommander.Parameter;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+
+import com.google.common.hash.Hashing;
+import com.google.common.hash.Hasher;
+
+
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.concurrent.CompletableFuture;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+import org.apache.bookkeeper.client.*;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.LedgerHandleAdv;
+import org.apache.bookkeeper.client.api.BookKeeper;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.client.api.LedgerEntries;
+import org.apache.bookkeeper.client.api.LedgerEntry;
+import org.apache.bookkeeper.client.api.LedgerMetadata;
+import org.apache.bookkeeper.meta.LedgerMetadataSerDe;
+import org.apache.bookkeeper.versioning.Versioned;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.tools.cli.helpers.ClientCommand;
+import org.apache.bookkeeper.tools.framework.CliFlags;
+import org.apache.bookkeeper.tools.framework.CliSpec;
+import org.apache.bookkeeper.util.IOUtils;
+import org.apache.bookkeeper.util.LedgerIdFormatter;
+import io.netty.buffer.ByteBufAllocator;
+
+import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.bookkeeper.common.util.OrderedScheduler;
+import org.apache.bookkeeper.meta.LedgerManager;
+import org.apache.bookkeeper.proto.BookieClient;
+
+/**
+ * Command to recopy a given ledger.
+ */
+public class GenerateChecksums extends ClientCommand<GenerateChecksums.Flags> {
+
+ private static final String NAME = "generate-checksums";
+ private static final String DESC = "Read each ledger, generate checksums and store to file";
+ private static final String DEFAULT = "";
+
+ public GenerateChecksums() {
+ this(new Flags());
+ }
+
+ private GenerateChecksums(Flags flags) {
+ super(CliSpec.<GenerateChecksums.Flags>newBuilder()
+ .withName(NAME)
+ .withDescription(DESC)
+ .withFlags(flags)
+ .build());
+ }
+
+ @Accessors(fluent = true)
+ @Setter
+ public static class Flags extends CliFlags {
+ @Parameter(names = { "--batch-size" }, description = "Number of entries to copy at a time")
+ private long batchSize = 1000;
+ }
+
+ @Override
+ protected void run(BookKeeper bk, Flags flags) throws Exception {
+ try (BookKeeperAdmin admin = new BookKeeperAdmin((org.apache.bookkeeper.client.BookKeeper) bk)) {
+ for (long lid : admin.listLedgers()) {
+ try (FileOutputStream f = new FileOutputStream("ledger-" + lid + ".murmur3")) {
+ f.write(generateChecksum(admin, lid, flags.batchSize));
+ }
+ System.out.println("Checksum generated for " + lid);
+ }
+ } catch (Throwable t) {
+ t.printStackTrace();
+ throw new UncheckedExecutionException(t.getMessage(), t);
+ }
+ }
+
+ static byte[] generateChecksum(BookKeeperAdmin admin, long ledgerId, long batchSize)
+ throws Exception {
+ CompletableFuture<LedgerHandle> promise = new CompletableFuture<>();
+ admin.asyncOpenLedger(ledgerId, (rc, lh, ignore) -> {
+ if (rc != BKException.Code.OK) {
+ promise.completeExceptionally(BKException.create(rc));
+ } else {
+ promise.complete(lh);
+ }
+ }, null);
+ Hasher hasher = Hashing.murmur3_128().newHasher();
+ try (LedgerHandle rh = promise.get()) {
+ hasher.putString(rh.getLedgerMetadata().getDigestType().toString(), StandardCharsets.UTF_8);
+ hasher.putBytes(rh.getLedgerMetadata().getPassword());
+ for (long eid = 0; eid <= rh.getLastAddConfirmed(); eid += batchSize) {
+ try (LedgerEntries entries = rh.read(eid, Math.min(eid + (batchSize - 1), rh.getLastAddConfirmed()))) {
+ for (LedgerEntry e : entries) {
+ hasher.putBytes(e.getEntryBytes());
+ hasher.putLong(e.getLength());
+ }
+ }
+ }
+ }
+ return hasher.hash().asBytes();
+ }
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/client/LedgerRewriter.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/client/LedgerRewriter.java
new file mode 100644
index 0000000..4147562
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/client/LedgerRewriter.java
@@ -0,0 +1,73 @@
+package org.apache.bookkeeper.tools.cli.commands.client;
+
+import static org.apache.bookkeeper.proto.BookieProtocol.FLAG_RECOVERY_ADD;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.util.ReferenceCountUtil;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.bookkeeper.client.api.LedgerMetadata;
+import org.apache.bookkeeper.client.api.WriteFlag;
+import org.apache.bookkeeper.client.DistributionSchedule;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.proto.BookieClient;
+import org.apache.bookkeeper.proto.checksum.DigestManager;
+import org.apache.bookkeeper.util.ByteBufList;
+
+public class LedgerRewriter {
+ private final long ledgerId;
+ private final LedgerMetadata metadata;
+ private final DistributionSchedule distSchedule;
+ private final DigestManager digests;
+ private final List<BookieSocketAddress> ensemble;
+ private final BookieClient bookieClient;
+
+ public LedgerRewriter(long ledgerId, LedgerMetadata metadata,
+ DistributionSchedule distSchedule,
+ DigestManager digests,
+ BookieClient bookieClient) {
+ this.ledgerId = ledgerId;
+ this.metadata = metadata;
+ this.distSchedule = distSchedule;
+ this.digests = digests;
+ this.ensemble = metadata.getAllEnsembles().get(0L);
+ this.bookieClient = bookieClient;
+ }
+
+ public CompletableFuture<Void> write(long entryId, long lac,
+ long length,
+ ByteBuf buffer) {
+ DistributionSchedule.WriteSet writeSet = distSchedule.getWriteSet(entryId);
+
+ ByteBufList toSend = digests.computeDigestAndPackageForSending(
+ entryId, lac, length, buffer);
+ try {
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ for (int i = 0; i < writeSet.size(); i++) {
+ BookieSocketAddress bookie = ensemble.get(writeSet.get(i));
+ CompletableFuture<Void> bookieAddFuture = new CompletableFuture<>();
+ futures.add(bookieAddFuture);
+ bookieClient.addEntry(bookie, ledgerId,
+ metadata.getPassword(), entryId, toSend,
+ (rc, ledgerId, entryId2, addr, ctx) -> {
+ if (rc != BKException.Code.OK) {
+ bookieAddFuture.completeExceptionally(
+ BKException.create(rc));
+ } else {
+ bookieAddFuture.complete(null);
+ }
+ }, null, FLAG_RECOVERY_ADD, false, WriteFlag.NONE);
+ }
+ return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
+ .whenComplete((ignore, exception) -> {
+ ReferenceCountUtil.release(toSend);
+ });
+ } finally {
+ writeSet.recycle();
+ }
+ }
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/client/RecopyLedgerCommand.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/client/RecopyLedgerCommand.java
new file mode 100644
index 0000000..ca880ed
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/client/RecopyLedgerCommand.java
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.tools.cli.commands.client;
+import org.apache.bookkeeper.client.api.WriteFlag;
+
+import com.beust.jcommander.Parameter;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.concurrent.CompletableFuture;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+import org.apache.bookkeeper.client.*;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.LedgerHandleAdv;
+import org.apache.bookkeeper.client.api.BookKeeper;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.client.api.LedgerEntries;
+import org.apache.bookkeeper.client.api.LedgerEntry;
+import org.apache.bookkeeper.client.api.LedgerMetadata;
+import org.apache.bookkeeper.meta.LedgerMetadataSerDe;
+import org.apache.bookkeeper.versioning.Versioned;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.tools.cli.helpers.ClientCommand;
+import org.apache.bookkeeper.tools.framework.CliFlags;
+import org.apache.bookkeeper.tools.framework.CliSpec;
+import org.apache.bookkeeper.util.IOUtils;
+import org.apache.bookkeeper.util.LedgerIdFormatter;
+import io.netty.buffer.ByteBufAllocator;
+
+import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.bookkeeper.common.util.OrderedScheduler;
+import org.apache.bookkeeper.meta.LedgerManager;
+import org.apache.bookkeeper.proto.BookieClient;
+
+/**
+ * Command to recopy a given ledger.
+ */
+public class RecopyLedgerCommand extends ClientCommand<RecopyLedgerCommand.Flags> {
+
+ private static final String NAME = "recopy";
+ private static final String DESC = "Read and rewrite a ledger, optionally avoiding some bookies";
+ private static final String DEFAULT = "";
+
+ private LedgerIdFormatter ledgerIdFormatter;
+
+ public RecopyLedgerCommand() {
+ this(new Flags());
+ }
+
+ private RecopyLedgerCommand(Flags flags) {
+ super(CliSpec.<RecopyLedgerCommand.Flags>newBuilder()
+ .withName(NAME)
+ .withDescription(DESC)
+ .withFlags(flags)
+ .build());
+ }
+
+ @Accessors(fluent = true)
+ @Setter
+ public static class Flags extends CliFlags {
+
+ @Parameter(names = { "-l", "--ledgerid" }, description = "Ledger ID")
+ private long ledgerId;
+
+ @Parameter(names = { "-a", "--all" }, description = "Recopy all ledgers")
+ private boolean all = false;
+
+ @Parameter(names = { "--avoid-bookies" }, description = "Avoid writing to a set of bookies (comma separated)")
+ private String avoid = "";
+
+ @Parameter(names = { "--batch-size" }, description = "Number of entries to copy at a time")
+ private long batchSize = 1000;
+ }
+
+ @Override
+ protected void run(BookKeeper bk, Flags flags) throws Exception {
+ try (BookKeeperAdmin admin = new BookKeeperAdmin((org.apache.bookkeeper.client.BookKeeper) bk)) {
+ Set<BookieSocketAddress> avoid = new HashSet<>();
+ for (String s : flags.avoid.split(",")) {
+ avoid.add(new BookieSocketAddress(s));
+ }
+ if (flags.all) {
+ if (avoid.isEmpty()) {
+ System.out.println("No bookies to avoid, so no ledgers copied");
+ return;
+ }
+ ClientContext ctx = ((org.apache.bookkeeper.client.BookKeeper) bk).getClientCtx();
+ LedgerManager lm = ctx.getLedgerManager();
+ for (long lid : admin.listLedgers()) {
+ boolean hasBookie = lm.readLedgerMetadata(lid).get()
+ .getValue().getAllEnsembles()
+ .values().stream()
+ .flatMap(ensemble -> ensemble.stream())
+ .anyMatch((b) -> avoid.contains(b));
+ if (hasBookie) {
+ System.out.println("Ledger " + lid + " has bookie in avoidSet, copying");
+ recopyLedger(bk, admin, lid, avoid, flags.batchSize);
+ }
+ }
+ } else if (flags.ledgerId != -1) {
+ recopyLedger(bk, admin, flags.ledgerId, avoid, flags.batchSize);
+ } else {
+ throw new Exception("Must specify a ledger id or all");
+ }
+ } catch (Throwable t) {
+ t.printStackTrace();
+ throw new UncheckedExecutionException(t.getMessage(), t);
+ }
+ }
+
+ private void recopyLedger(BookKeeper bk, BookKeeperAdmin admin, long ledgerId, Set<BookieSocketAddress> avoid,
+ long batchSize)
+ throws Exception {
+ CompletableFuture<LedgerHandle> promise = new CompletableFuture<>();
+ admin.asyncOpenLedger(ledgerId, (rc, lh, ignore) -> {
+ if (rc != BKException.Code.OK) {
+ promise.completeExceptionally(BKException.create(rc));
+ } else {
+ promise.complete(lh);
+ }
+ }, null);
+ ClientContext ctx = ((org.apache.bookkeeper.client.BookKeeper) bk).getClientCtx();
+ LedgerManager lm = ctx.getLedgerManager();
+ LedgerMetadataSerDe serDe = new LedgerMetadataSerDe();
+ try (LedgerHandle rh = promise.get()) {
+ Versioned<LedgerMetadata> origVersionedMetadata = rh.getVersionedLedgerMetadata();
+ LedgerMetadata origMetadata = origVersionedMetadata.getValue();
+
+ try (FileOutputStream outf = new FileOutputStream(String.valueOf(rh.getId()) + ".backup.bin")) {
+ outf.write(serDe.serialize(origMetadata));
+ }
+
+ List<BookieSocketAddress> newEnsemble = ctx.getPlacementPolicy().newEnsemble(
+ origMetadata.getEnsembleSize(), origMetadata.getWriteQuorumSize(),
+ origMetadata.getAckQuorumSize(), origMetadata.getCustomMetadata(),
+ avoid).getResult();
+ LedgerMetadata newMetadata = LedgerMetadataBuilder.from(origMetadata)
+ .clearEnsembles()
+ .newEnsembleEntry(0L, newEnsemble)
+ .build();
+ LedgerRewriter rewriter = new LedgerRewriter(ledgerId, newMetadata,
+ rh.getDistributionSchedule(),
+ rh.getDigestManager(),
+ ctx.getBookieClient());
+ long lac = -1;
+ long i = 0;
+
+ for (long eid = 0; eid <= rh.getLastAddConfirmed(); eid += batchSize) {
+ try (LedgerEntries entries = rh.read(eid, Math.min(eid + (batchSize - 1), rh.getLastAddConfirmed()))) {
+ List<CompletableFuture<Void>> futures= new ArrayList<>();
+ for (LedgerEntry e : entries) {
+ futures.add(rewriter.write(e.getEntryId(),
+ lac, e.getLength(),
+ e.getEntryBuffer().retain()));
+ lac = e.getEntryId();
+ i++;
+ }
+ CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();
+ }
+ }
+ lm.writeLedgerMetadata(rh.getId(), newMetadata,
+ origVersionedMetadata.getVersion());
+ System.out.println("Rewrote " + i + " entries for ledger " + rh.getId());
+ }
+ }
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/client/ReferencedBookies.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/client/ReferencedBookies.java
new file mode 100644
index 0000000..6e27b26
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/client/ReferencedBookies.java
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.tools.cli.commands.client;
+import org.apache.bookkeeper.client.api.WriteFlag;
+
+import com.beust.jcommander.Parameter;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+
+import com.google.common.hash.Hashing;
+import com.google.common.hash.Hasher;
+
+
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.concurrent.CompletableFuture;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+import org.apache.bookkeeper.client.*;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.LedgerHandleAdv;
+import org.apache.bookkeeper.client.api.BookKeeper;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.client.api.LedgerEntries;
+import org.apache.bookkeeper.client.api.LedgerEntry;
+import org.apache.bookkeeper.client.api.LedgerMetadata;
+import org.apache.bookkeeper.meta.LedgerMetadataSerDe;
+import org.apache.bookkeeper.versioning.Versioned;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.tools.cli.helpers.ClientCommand;
+import org.apache.bookkeeper.tools.framework.CliFlags;
+import org.apache.bookkeeper.tools.framework.CliSpec;
+import org.apache.bookkeeper.util.IOUtils;
+import org.apache.bookkeeper.util.LedgerIdFormatter;
+import io.netty.buffer.ByteBufAllocator;
+
+import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.bookkeeper.common.util.OrderedScheduler;
+import org.apache.bookkeeper.meta.LedgerManager;
+import org.apache.bookkeeper.proto.BookieClient;
+
+/**
+ * Command to recopy a given ledger.
+ */
+public class ReferencedBookies extends ClientCommand<CliFlags> {
+
+ private static final String NAME = "referenced-bookies";
+ private static final String DESC = "List all the bookies referenced by ledgers";
+ private static final String DEFAULT = "";
+
+ public ReferencedBookies() {
+ this(new CliFlags());
+ }
+
+ private ReferencedBookies(CliFlags flags) {
+ super(CliSpec.<CliFlags>newBuilder()
+ .withName(NAME)
+ .withDescription(DESC)
+ .withFlags(flags)
+ .build());
+ }
+
+ @Override
+ protected void run(BookKeeper bk, CliFlags flags) throws Exception {
+ Set<BookieSocketAddress> referencedBookies = new HashSet<>();
+ try (BookKeeperAdmin admin = new BookKeeperAdmin((org.apache.bookkeeper.client.BookKeeper) bk)) {
+ for (long lid : admin.listLedgers()) {
+ referencedBookies.addAll(getReferencedBookies(admin, lid));
+ }
+ int i = 1;
+ for (BookieSocketAddress a : referencedBookies) {
+ System.out.println(String.format("%d: %s", i++, a));
+ }
+ } catch (Throwable t) {
+ t.printStackTrace();
+ throw new UncheckedExecutionException(t.getMessage(), t);
+ }
+ }
+
+ static Set<BookieSocketAddress> getReferencedBookies(BookKeeperAdmin admin, long ledgerId)
+ throws Exception {
+ Set<BookieSocketAddress> bookies = new HashSet<>();
+ try (LedgerHandle lh = admin.openLedger(ledgerId)) {
+ for (List<BookieSocketAddress> ensemble : lh.getLedgerMetadata().getAllEnsembles().values()) {
+ bookies.addAll(ensemble);
+ }
+ }
+ return bookies;
+ }
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/client/VerifyChecksums.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/client/VerifyChecksums.java
new file mode 100644
index 0000000..835f586
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/client/VerifyChecksums.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.tools.cli.commands.client;
+import org.apache.bookkeeper.client.api.WriteFlag;
+
+import com.beust.jcommander.Parameter;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+
+import com.google.common.hash.Hashing;
+import com.google.common.hash.Hasher;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.concurrent.CompletableFuture;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+import org.apache.bookkeeper.client.*;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.LedgerHandleAdv;
+import org.apache.bookkeeper.client.api.BookKeeper;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.client.api.LedgerEntries;
+import org.apache.bookkeeper.client.api.LedgerEntry;
+import org.apache.bookkeeper.client.api.LedgerMetadata;
+import org.apache.bookkeeper.meta.LedgerMetadataSerDe;
+import org.apache.bookkeeper.versioning.Versioned;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.tools.cli.helpers.ClientCommand;
+import org.apache.bookkeeper.tools.framework.CliFlags;
+import org.apache.bookkeeper.tools.framework.CliSpec;
+import org.apache.bookkeeper.util.IOUtils;
+import org.apache.bookkeeper.util.LedgerIdFormatter;
+import io.netty.buffer.ByteBufAllocator;
+
+import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.bookkeeper.common.util.OrderedScheduler;
+import org.apache.bookkeeper.meta.LedgerManager;
+import org.apache.bookkeeper.proto.BookieClient;
+
+/**
+ * Command to recopy a given ledger.
+ */
+public class VerifyChecksums extends ClientCommand<VerifyChecksums.Flags> {
+
+ private static final String NAME = "verify-checksums";
+ private static final String DESC = "Verify generated checksums match what can be read from the cluster";
+ private static final String DEFAULT = "";
+
+ public VerifyChecksums() {
+ this(new Flags());
+ }
+
+ private VerifyChecksums(Flags flags) {
+ super(CliSpec.<VerifyChecksums.Flags>newBuilder()
+ .withName(NAME)
+ .withDescription(DESC)
+ .withFlags(flags)
+ .build());
+ }
+
+ @Accessors(fluent = true)
+ @Setter
+ public static class Flags extends CliFlags {
+ @Parameter(names = { "--batch-size" }, description = "Number of entries to copy at a time")
+ private long batchSize = 1000;
+ }
+
+ @Override
+ protected void run(BookKeeper bk, Flags flags) throws Exception {
+ try (BookKeeperAdmin admin = new BookKeeperAdmin((org.apache.bookkeeper.client.BookKeeper) bk)) {
+ for (File file : new File(".").listFiles(
+ (dir, name) -> name.startsWith("ledger-") && name.endsWith(".murmur3"))) {
+ long lid = Long.valueOf(file.getName().replace("ledger-", "").replace(".murmur3", ""));
+ byte[] previous = Files.readAllBytes(file.toPath());
+ byte[] current = GenerateChecksums.generateChecksum(admin, lid, flags.batchSize);
+
+ if (Arrays.equals(previous, current)) {
+ System.out.println("Checksum good for ledger " + lid);
+ } else {
+ System.out.println("Checksum BAD for ledger " + lid);
+ throw new Exception("Bad checksum for " + lid);
+ }
+ }
+ } catch (Throwable t) {
+ t.printStackTrace();
+ throw new UncheckedExecutionException(t.getMessage(), t);
+ }
+ }
+}
diff --git a/tests/integration/smoke/src/test/java/org/apache/bookkeeper/tests/integration/TestCLI.java b/tests/integration/smoke/src/test/java/org/apache/bookkeeper/tests/integration/TestCLI.java
index 781d28e..c53204d 100644
--- a/tests/integration/smoke/src/test/java/org/apache/bookkeeper/tests/integration/TestCLI.java
+++ b/tests/integration/smoke/src/test/java/org/apache/bookkeeper/tests/integration/TestCLI.java
@@ -26,6 +26,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.api.DigestType;
+import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.client.api.WriteHandle;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.tests.integration.utils.BookKeeperClusterUtils;
@@ -72,83 +73,119 @@ public class TestCLI {
assertTrue(BookKeeperClusterUtils.stopAllBookies(docker));
}
- @Test
- public void test001_SimpleTest() throws Exception {
- String bookie = BookKeeperClusterUtils.getAnyBookie();
- assertTrue(DockerUtils.runCommand(docker, bookie,
- bkctl,
- "ledger",
- "simpletest",
- "--ensemble-size", "3",
- "--write-quorum-size", "3",
- "--ack-quorum-size", "2",
- "--num-entries", "100"
- ).contains("100 entries written to ledger"));
- }
-
- @Test
- public void test002_ListROBookies() throws Exception {
- String bookie = BookKeeperClusterUtils.getAnyBookie();
- assertTrue(DockerUtils.runCommand(docker, bookie,
- bkctl,
- "bookies",
- "list",
- "-ro"
- ).contains("No bookie exists!"));
- }
-
- @Test
- public void test003_ListRWBookies() throws Exception {
- String bookie = BookKeeperClusterUtils.getAnyBookie();
- assertTrue(DockerUtils.runCommand(docker, bookie,
- bkctl,
- "bookies",
- "list",
- "-rw"
- ).contains("ReadWrite Bookies :"));
- }
+ // @Test
+ // public void test001_SimpleTest() throws Exception {
+ // String bookie = BookKeeperClusterUtils.getAnyBookie();
+ // assertTrue(DockerUtils.runCommand(docker, bookie,
+ // bkctl,
+ // "ledger",
+ // "simpletest",
+ // "--ensemble-size", "3",
+ // "--write-quorum-size", "3",
+ // "--ack-quorum-size", "2",
+ // "--num-entries", "100"
+ // ).contains("100 entries written to ledger"));
+ // }
+
+ // @Test
+ // public void test002_ListROBookies() throws Exception {
+ // String bookie = BookKeeperClusterUtils.getAnyBookie();
+ // assertTrue(DockerUtils.runCommand(docker, bookie,
+ // bkctl,
+ // "bookies",
+ // "list",
+ // "-ro"
+ // ).contains("No bookie exists!"));
+ // }
+
+ // @Test
+ // public void test003_ListRWBookies() throws Exception {
+ // String bookie = BookKeeperClusterUtils.getAnyBookie();
+ // assertTrue(DockerUtils.runCommand(docker, bookie,
+ // bkctl,
+ // "bookies",
+ // "list",
+ // "-rw"
+ // ).contains("ReadWrite Bookies :"));
+ // }
+
+ // @Test
+ // public void test004_SearchReplaceBookieId() throws Exception {
+ // String zookeeper = BookKeeperClusterUtils.zookeeperConnectString(docker);
+
+ // String bookie = BookKeeperClusterUtils.getAnyBookie();
+ // int numEntries = 100;
+ // try (BookKeeper bk = new BookKeeper(zookeeper)) {
+ // long ledgerId;
+ // BookieSocketAddress toReplace;
+ // BookieSocketAddress replaceWith = new BookieSocketAddress("192.0.2.1:3181");
+ // try (WriteHandle writelh = bk.newCreateLedgerOp()
+ // .withDigestType(DigestType.CRC32C).withPassword(TestSmoke.PASSWD)
+ // .withEnsembleSize(1).withWriteQuorumSize(1).withAckQuorumSize(1).execute().get()) {
+ // ledgerId = writelh.getId();
+ // toReplace = writelh.getLedgerMetadata().getAllEnsembles().get(0L).get(0);
+ // for (int i = 0; i < numEntries; i++) {
+ // writelh.append(("entry-" + i).getBytes());
+ // }
+ // }
+
+ // TestSmoke.readEntries(bk, ledgerId, numEntries);
+
+ // DockerUtils.runCommand(docker, bookie,
+ // bkctl,
+ // "bookieid", "searchreplace",
+ // "--from", toReplace.toString(),
+ // "--to", replaceWith.toString());
+
+ // try {
+ // TestSmoke.readEntries(bk, ledgerId, numEntries);
+ // fail("Shouldn't be able to read, as bookie id is rubbish");
+ // } catch (BKException.BKBookieHandleNotAvailableException e) {
+ // // expected
+ // }
+
+ // DockerUtils.runCommand(docker, bookie,
+ // bkctl,
+ // "bookieid", "searchreplace",
+ // "--from", replaceWith.toString(),
+ // "--to", toReplace.toString());
+ // TestSmoke.readEntries(bk, ledgerId, numEntries);
+ // }
+ // }
@Test
- public void test004_SearchReplaceBookieId() throws Exception {
+ public void test005_RecopyLedger() throws Exception {
String zookeeper = BookKeeperClusterUtils.zookeeperConnectString(docker);
String bookie = BookKeeperClusterUtils.getAnyBookie();
int numEntries = 100;
+
try (BookKeeper bk = new BookKeeper(zookeeper)) {
long ledgerId;
- BookieSocketAddress toReplace;
- BookieSocketAddress replaceWith = new BookieSocketAddress("192.0.2.1:3181");
+ BookieSocketAddress toAvoid = null;
try (WriteHandle writelh = bk.newCreateLedgerOp()
.withDigestType(DigestType.CRC32C).withPassword(TestSmoke.PASSWD)
.withEnsembleSize(1).withWriteQuorumSize(1).withAckQuorumSize(1).execute().get()) {
ledgerId = writelh.getId();
- toReplace = writelh.getLedgerMetadata().getAllEnsembles().get(0L).get(0);
+ toAvoid = writelh.getLedgerMetadata().getAllEnsembles().get(0L).get(0);
for (int i = 0; i < numEntries; i++) {
writelh.append(("entry-" + i).getBytes());
}
}
-
TestSmoke.readEntries(bk, ledgerId, numEntries);
DockerUtils.runCommand(docker, bookie,
bkctl,
- "bookieid", "searchreplace",
- "--from", toReplace.toString(),
- "--to", replaceWith.toString());
-
- try {
- TestSmoke.readEntries(bk, ledgerId, numEntries);
- fail("Shouldn't be able to read, as bookie id is rubbish");
- } catch (BKException.BKBookieHandleNotAvailableException e) {
- // expected
- }
-
- DockerUtils.runCommand(docker, bookie,
- bkctl,
- "bookieid", "searchreplace",
- "--from", replaceWith.toString(),
- "--to", toReplace.toString());
+ "ledger", "recopy",
+ "--all",
+ "--avoid-bookies", toAvoid.toString());
TestSmoke.readEntries(bk, ledgerId, numEntries);
+ try (ReadHandle readLh = bk.newOpenLedgerOp()
+ .withDigestType(DigestType.CRC32C).withPassword(TestSmoke.PASSWD)
+ .withLedgerId(ledgerId).execute().get()) {
+ assertTrue(!readLh.getLedgerMetadata().getAllEnsembles().get(0L).contains(toAvoid));
+ }
}
}
+
}
diff --git a/tools/ledger/src/main/java/org/apache/bookkeeper/tools/cli/commands/LedgerCommandGroup.java b/tools/ledger/src/main/java/org/apache/bookkeeper/tools/cli/commands/LedgerCommandGroup.java
index a133b5d..e9c0bbf 100644
--- a/tools/ledger/src/main/java/org/apache/bookkeeper/tools/cli/commands/LedgerCommandGroup.java
+++ b/tools/ledger/src/main/java/org/apache/bookkeeper/tools/cli/commands/LedgerCommandGroup.java
@@ -23,7 +23,11 @@ import static org.apache.bookkeeper.tools.common.BKCommandCategories.CATEGORY_LE
import org.apache.bookkeeper.tools.cli.BKCtl;
import org.apache.bookkeeper.tools.cli.commands.client.DeleteLedgerCommand;
import org.apache.bookkeeper.tools.cli.commands.client.LedgerMetaDataCommand;
+import org.apache.bookkeeper.tools.cli.commands.client.RecopyLedgerCommand;
import org.apache.bookkeeper.tools.cli.commands.client.SimpleTestCommand;
+import org.apache.bookkeeper.tools.cli.commands.client.GenerateChecksums;
+import org.apache.bookkeeper.tools.cli.commands.client.VerifyChecksums;
+import org.apache.bookkeeper.tools.cli.commands.client.ReferencedBookies;
import org.apache.bookkeeper.tools.common.BKFlags;
import org.apache.bookkeeper.tools.framework.CliCommandGroup;
import org.apache.bookkeeper.tools.framework.CliSpec;
@@ -44,6 +48,10 @@ public class LedgerCommandGroup extends CliCommandGroup<BKFlags> {
.addCommand(new SimpleTestCommand())
.addCommand(new DeleteLedgerCommand())
.addCommand(new LedgerMetaDataCommand())
+ .addCommand(new RecopyLedgerCommand())
+ .addCommand(new GenerateChecksums())
+ .addCommand(new VerifyChecksums())
+ .addCommand(new ReferencedBookies())
.build();
public LedgerCommandGroup() {