You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bookkeeper.apache.org by GitBox <gi...@apache.org> on 2022/09/13 10:30:29 UTC
[GitHub] [bookkeeper] eolivelli commented on a diff in pull request #3457: BP-56: Support data migration
eolivelli commented on code in PR #3457:
URL: https://github.com/apache/bookkeeper/pull/3457#discussion_r969445823
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/bookies/ReplicasMigrationCommand.java:
##########
@@ -0,0 +1,190 @@
+/*
+ * 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.bookies;
+
+import com.beust.jcommander.Parameter;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.http.HttpRouter;
+import org.apache.bookkeeper.http.HttpServer;
+import org.apache.bookkeeper.meta.MetadataDrivers;
+import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.tools.cli.helpers.BookieCommand;
+import org.apache.bookkeeper.tools.framework.CliFlags;
+import org.apache.bookkeeper.tools.framework.CliSpec;
+import org.apache.bookkeeper.util.HttpClientUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Migrate the data of the specified replica to other bookie nodes.
+ * */
+public class ReplicasMigrationCommand extends BookieCommand<ReplicasMigrationCommand.ReplicasMigrationFlags> {
+ static final Logger LOG = LoggerFactory.getLogger(ReplicasMigrationCommand.class);
+ private static final String NAME = "replicasMigration";
+ private static final String DESC = "Migrate the data of the specified replica to other bookie nodes";
+ private static final String ALL = "ALL";
+ private final String seperator = ",";
+ public ReplicasMigrationCommand() {
+ this(new ReplicasMigrationFlags());
+ }
+ private ReplicasMigrationCommand(ReplicasMigrationFlags flags) {
+ super(CliSpec.<ReplicasMigrationFlags>newBuilder()
+ .withName(NAME)
+ .withDescription(DESC)
+ .withFlags(flags)
+ .build());
+ }
+
+ /**
+ * Flags for replicasMigration command.
+ */
+ @Accessors(fluent = true)
+ @Setter
+ public static class ReplicasMigrationFlags extends CliFlags {
+
+ @Parameter(names = { "-b", "--bookieIds" }, description = "Bookies corresponding to the migrated replica, "
+ + "eg: bookieId1,bookieId2,bookieId3")
+ private String bookieIdsOfMigratedReplica;
+
+ @Parameter(names = { "-l", "--ledgerIds" }, description = "The ledgerIds corresponding to the migrated replica,"
+ + "eg: ledgerId1,ledgerId2,ledgerId3. The default is ALL, which means that all replica data on these "
+ + "bookie nodes will be migrated to other bookie nodes")
+ private String ledgerIdsOfMigratedReplica;
+
+ @Parameter(names = { "-r", "--readOnly" }, description = "Whether to set the bookie nodes of the migrated data"
+ + " to readOnly, the default is true")
+ private boolean readOnly;
+
+ @Parameter(names = {"-p", "--httpPort"}, description = "http port,default is 8080")
+ private String port;
+ }
+
+ @Override
+ public boolean apply(ServerConfiguration conf, ReplicasMigrationFlags cmdFlags) {
+ try {
+ return switchToReadonly(conf, cmdFlags) && replicasMigration(conf, cmdFlags);
+ } catch (Exception e) {
+ throw new UncheckedExecutionException(e.getMessage(), e);
+ }
+ }
+
+ public Set<BookieId> parseBookieIds(String bookieIdsOfMigratedReplica) {
+ Set<BookieId> bookieIdsSet = new HashSet<>();
+ String[] bookieIds = bookieIdsOfMigratedReplica.split(seperator);
+ for (String bookieId : bookieIds) {
+ bookieIdsSet.add(BookieId.parse(bookieId));
+ }
+ return bookieIdsSet;
+ }
+
+ public Set<Long> parseLedgerIds(String ledgerIdsOfMigratedReplica) {
+ Set<Long> ledgerIdSet = new HashSet<>();
+ String[] ledgerIds = ledgerIdsOfMigratedReplica.split(seperator);
+ for (String ledgerId : ledgerIds) {
+ ledgerIdSet.add(Long.parseLong(ledgerId));
+ }
+ return ledgerIdSet;
+ }
+
+ private boolean switchToReadonly(ServerConfiguration conf, ReplicasMigrationFlags flags) {
+ if (flags.readOnly && !conf.isHttpServerEnabled()) {
+ throw new RuntimeException("Please enable http service first, config httpServerEnabled is false!");
+ }
+
+ Set<BookieId> bookieIdSet = parseBookieIds(flags.bookieIdsOfMigratedReplica);
+ for (BookieId bookieId : bookieIdSet) {
+ String bookieAddress = bookieId.getId().split(":")[0];
Review Comment:
this doesn't work if your bookieId is not a public hostname.
the bookieId is a opaque string, we have to "resolve" the ID
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/util/HttpClientUtils.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * 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.util;
+
+import java.io.IOException;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClientBuilder;
+
+/**
+ * http client utils.
+ * */
+public class HttpClientUtils {
+ private static final CloseableHttpClient httpClient = HttpClientBuilder.create().build();
Review Comment:
we should close this object
using a static field is not good.
we can open/close the client per each request, it is not a big deal
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MetadataClientDriver.java:
##########
@@ -106,6 +110,27 @@ interface SessionStateListener {
*/
void setSessionStateListener(SessionStateListener sessionStateListener);
+ default List<String> listLedgersOfMigrationReplicas(String migrationReplicasPath)
+ throws InterruptedException, KeeperException {
+ return new ArrayList<>();
+ }
+
+ /**
+ * After obtaining a replica migration task,
+ * lock the replica task to prevent it from being executed by other workers.
+ * */
+ default void lockMigrationReplicas(String lockPath, String advertisedAddress)
+ throws InterruptedException, KeeperException {
+ }
+
+ default String getOwnerBookiesMigrationReplicas(String ledgerForMigrationReplicasPath)
+ throws InterruptedException, KeeperException, UnsupportedEncodingException {
+ return "";
+ }
+
+ default void deleteZkPath(String migrationLedgerPath) throws InterruptedException, KeeperException {
+ }
Review Comment:
I would throw some UnsupportedOperationException
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MetadataClientDriver.java:
##########
@@ -106,6 +110,27 @@ interface SessionStateListener {
*/
void setSessionStateListener(SessionStateListener sessionStateListener);
+ default List<String> listLedgersOfMigrationReplicas(String migrationReplicasPath)
+ throws InterruptedException, KeeperException {
+ return new ArrayList<>();
+ }
+
+ /**
+ * After obtaining a replica migration task,
+ * lock the replica task to prevent it from being executed by other workers.
+ * */
+ default void lockMigrationReplicas(String lockPath, String advertisedAddress)
+ throws InterruptedException, KeeperException {
+ }
+
+ default String getOwnerBookiesMigrationReplicas(String ledgerForMigrationReplicasPath)
+ throws InterruptedException, KeeperException, UnsupportedEncodingException {
+ return "";
Review Comment:
I would throw some UnsupportedOperationException
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/bookies/ReplicasMigrationCommand.java:
##########
@@ -0,0 +1,190 @@
+/*
+ * 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.bookies;
+
+import com.beust.jcommander.Parameter;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.http.HttpRouter;
+import org.apache.bookkeeper.http.HttpServer;
+import org.apache.bookkeeper.meta.MetadataDrivers;
+import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.tools.cli.helpers.BookieCommand;
+import org.apache.bookkeeper.tools.framework.CliFlags;
+import org.apache.bookkeeper.tools.framework.CliSpec;
+import org.apache.bookkeeper.util.HttpClientUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Migrate the data of the specified replica to other bookie nodes.
+ * */
+public class ReplicasMigrationCommand extends BookieCommand<ReplicasMigrationCommand.ReplicasMigrationFlags> {
+ static final Logger LOG = LoggerFactory.getLogger(ReplicasMigrationCommand.class);
+ private static final String NAME = "replicasMigration";
+ private static final String DESC = "Migrate the data of the specified replica to other bookie nodes";
+ private static final String ALL = "ALL";
+ private final String seperator = ",";
+ public ReplicasMigrationCommand() {
+ this(new ReplicasMigrationFlags());
+ }
+ private ReplicasMigrationCommand(ReplicasMigrationFlags flags) {
+ super(CliSpec.<ReplicasMigrationFlags>newBuilder()
+ .withName(NAME)
+ .withDescription(DESC)
+ .withFlags(flags)
+ .build());
+ }
+
+ /**
+ * Flags for replicasMigration command.
+ */
+ @Accessors(fluent = true)
+ @Setter
+ public static class ReplicasMigrationFlags extends CliFlags {
+
+ @Parameter(names = { "-b", "--bookieIds" }, description = "Bookies corresponding to the migrated replica, "
+ + "eg: bookieId1,bookieId2,bookieId3")
+ private String bookieIdsOfMigratedReplica;
+
+ @Parameter(names = { "-l", "--ledgerIds" }, description = "The ledgerIds corresponding to the migrated replica,"
+ + "eg: ledgerId1,ledgerId2,ledgerId3. The default is ALL, which means that all replica data on these "
+ + "bookie nodes will be migrated to other bookie nodes")
+ private String ledgerIdsOfMigratedReplica;
+
+ @Parameter(names = { "-r", "--readOnly" }, description = "Whether to set the bookie nodes of the migrated data"
+ + " to readOnly, the default is true")
+ private boolean readOnly;
+
+ @Parameter(names = {"-p", "--httpPort"}, description = "http port,default is 8080")
+ private String port;
+ }
+
+ @Override
+ public boolean apply(ServerConfiguration conf, ReplicasMigrationFlags cmdFlags) {
+ try {
+ return switchToReadonly(conf, cmdFlags) && replicasMigration(conf, cmdFlags);
+ } catch (Exception e) {
+ throw new UncheckedExecutionException(e.getMessage(), e);
+ }
+ }
+
+ public Set<BookieId> parseBookieIds(String bookieIdsOfMigratedReplica) {
+ Set<BookieId> bookieIdsSet = new HashSet<>();
+ String[] bookieIds = bookieIdsOfMigratedReplica.split(seperator);
+ for (String bookieId : bookieIds) {
+ bookieIdsSet.add(BookieId.parse(bookieId));
+ }
+ return bookieIdsSet;
+ }
+
+ public Set<Long> parseLedgerIds(String ledgerIdsOfMigratedReplica) {
+ Set<Long> ledgerIdSet = new HashSet<>();
+ String[] ledgerIds = ledgerIdsOfMigratedReplica.split(seperator);
+ for (String ledgerId : ledgerIds) {
+ ledgerIdSet.add(Long.parseLong(ledgerId));
+ }
+ return ledgerIdSet;
+ }
+
+ private boolean switchToReadonly(ServerConfiguration conf, ReplicasMigrationFlags flags) {
+ if (flags.readOnly && !conf.isHttpServerEnabled()) {
+ throw new RuntimeException("Please enable http service first, config httpServerEnabled is false!");
+ }
+
+ Set<BookieId> bookieIdSet = parseBookieIds(flags.bookieIdsOfMigratedReplica);
+ for (BookieId bookieId : bookieIdSet) {
+ String bookieAddress = bookieId.getId().split(":")[0];
+ String httpAddress = bookieAddress + ":" + flags.port;
Review Comment:
if the bookie is up and running it publishes on metadata the http endpoint.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org