Added REST/CLI tools to manage bookies rack placement config (apache#…

…2005)

### Motivation

Currently we don't have tools to manage the rack placement configuration for bookies in a Pulsar cluster. 

### Modifications

 * Added `/admin/v2/bookie` REST handler
 * Added `admin.bookie()` in Java admin client
 * Added `pulsar-admin bookies ...` CLI tool 
 * Worked around an issue with bookie hostname and hostname:port that prevented BK client to detect the correct rack.

Fixes apache#151

pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java

@@ -38,11 +38,13 @@
 import org.apache.pulsar.common.policies.data.Policies;
 import org.apache.pulsar.common.policies.data.TenantInfo;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.apache.pulsar.zookeeper.ZkBookieRackAffinityMapping;
 import org.apache.pulsar.zookeeper.ZooKeeperClientFactory;
 import org.apache.pulsar.zookeeper.ZooKeeperClientFactory.SessionType;
 import org.apache.pulsar.zookeeper.ZookeeperClientFactoryImpl;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException.NodeExistsException;
+import org.apache.zookeeper.ZooDefs.Ids;
 import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.data.Stat;
@@ -136,6 +138,15 @@ public static void main(String[] args) throws Exception {
             throw new IOException("Failed to initialize BookKeeper metadata");
         }
 
+        if (localZk.exists(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, false) == null) {
+            try {
+                localZk.create(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, "{}".getBytes(), Ids.OPEN_ACL_UNSAFE,
+                        CreateMode.PERSISTENT);
+            } catch (NodeExistsException e) {
+                // Ignore
+            }
+        }
+
         localZk.create("/managed-ledgers", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
         localZk.create("/namespace", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Bookies.java

@@ -0,0 +1,142 @@
+/**
+ * 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.pulsar.broker.admin.v2;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import io.swagger.annotations.ApiResponse;
+import io.swagger.annotations.ApiResponses;
+
+import java.util.Map.Entry;
+import java.util.Optional;
+
+import javax.ws.rs.DELETE;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response.Status;
+
+import org.apache.pulsar.broker.admin.AdminResource;
+import org.apache.pulsar.broker.web.RestException;
+import org.apache.pulsar.common.policies.data.BookieInfo;
+import org.apache.pulsar.common.policies.data.BookiesRackConfiguration;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.apache.pulsar.zookeeper.ZkBookieRackAffinityMapping;
+import org.apache.pulsar.zookeeper.ZooKeeperCache.Deserializer;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Path("/bookies")
+@Api(value = "/bookies", description = "Configure bookies rack placement", tags = "bookies")
+@Produces(MediaType.APPLICATION_JSON)
+public class Bookies extends AdminResource {
+
+    @GET
+    @Path("/racks-info")
+    @ApiOperation(value = "Gets the rack placement information for all the bookies in the cluster", response = BookiesRackConfiguration.class)
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission") })
+    public BookiesRackConfiguration getBookiesRackInfo() throws Exception {
+        validateSuperUserAccess();
+
+        return localZkCache().getData(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH,
+                new Deserializer<BookiesRackConfiguration>() {
+
+                    @Override
+                    public BookiesRackConfiguration deserialize(String key, byte[] content) throws Exception {
+                        return ObjectMapperFactory.getThreadLocal().readValue(content, BookiesRackConfiguration.class);
+                    }
+
+                }).orElse(new BookiesRackConfiguration());
+    }
+
+    @GET
+    @Path("/racks-info/{bookie}")
+    @ApiOperation(value = "Gets the rack placement information for a specific bookie in the cluster", response = BookieInfo.class)
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission") })
+    public BookieInfo getBookieRackInfo(@PathParam("bookie") String bookieAddress) throws Exception {
+        validateSuperUserAccess();
+
+        BookiesRackConfiguration racks = localZkCache()
+                .getData(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, (key, content) -> ObjectMapperFactory
+                        .getThreadLocal().readValue(content, BookiesRackConfiguration.class))
+                .orElse(new BookiesRackConfiguration());
+
+        return racks.getBookie(bookieAddress)
+                .orElseThrow(() -> new RestException(Status.NOT_FOUND, "Bookie address not found: " + bookieAddress));
+    }
+
+    @DELETE
+    @Path("/racks-info/{bookie}")
+    @ApiOperation(value = "Removed the rack placement information for a specific bookie in the cluster")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission") })
+    public void deleteBookieRackInfo(@PathParam("bookie") String bookieAddress) throws Exception {
+        validateSuperUserAccess();
+
+        BookiesRackConfiguration racks = localZkCache()
+                .getData(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, (key, content) -> ObjectMapperFactory
+                        .getThreadLocal().readValue(content, BookiesRackConfiguration.class))
+                .orElse(new BookiesRackConfiguration());
+
+        if (!racks.removeBookie(bookieAddress)) {
+            throw new RestException(Status.NOT_FOUND, "Bookie address not found: " + bookieAddress);
+        }
+
+        log.info("Removed {} from rack mapping info", bookieAddress);
+    }
+
+    @POST
+    @Path("/racks-info/{bookie}")
+    @ApiOperation(value = "Updates the rack placement information for a specific bookie in the cluster")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission") })
+    public void updateBookieRackInfo(@PathParam("bookie") String bookieAddress, @QueryParam("group") String group,
+            BookieInfo bookieInfo) throws Exception {
+        validateSuperUserAccess();
+
+        if (group == null) {
+            throw new RestException(Status.PRECONDITION_FAILED, "Bookie 'group' parameters is missing");
+        }
+
+        Optional<Entry<BookiesRackConfiguration, Stat>> entry = localZkCache()
+                .getEntry(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, (key, content) -> ObjectMapperFactory
+                        .getThreadLocal().readValue(content, BookiesRackConfiguration.class));
+
+        if (entry.isPresent()) {
+            // Update the racks info
+            BookiesRackConfiguration racks = entry.get().getKey();
+            racks.updateBookie(group, bookieAddress, bookieInfo);
+
+            localZk().setData(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper().writeValueAsBytes(racks),
+                    entry.get().getValue().getVersion());
+            log.info("Updated rack mapping info for {}", bookieAddress);
+        } else {
+            // Creates the z-node with racks info
+            BookiesRackConfiguration racks = new BookiesRackConfiguration();
+            racks.updateBookie(group, bookieAddress, bookieInfo);
+            zkCreate(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper().writeValueAsBytes(racks));
+            log.info("Created rack mapping info and added {}", bookieAddress);
+        }
+    }
+
+    private static final Logger log = LoggerFactory.getLogger(Bookies.class);
+}

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java

@@ -71,11 +71,22 @@ public class BrokerBkEnsemblesTests {
     private final int ZOOKEEPER_PORT = 12759;
     protected final int BROKER_WEBSERVICE_PORT = 15782;
 
+    protected final int bkBasePort = 5001;
+    private final int numberOfBookies;
+
+    public BrokerBkEnsemblesTests() {
+        this(3);
+    }
+
+    public BrokerBkEnsemblesTests(int numberOfBookies) {
+        this.numberOfBookies = numberOfBookies;
+    }
+
     @BeforeMethod
     void setup() throws Exception {
         try {
             // start local bookie and zookeeper
-            bkEnsemble = new LocalBookkeeperEnsemble(3, ZOOKEEPER_PORT, 5001);
+            bkEnsemble = new LocalBookkeeperEnsemble(numberOfBookies, ZOOKEEPER_PORT, 5001);
             bkEnsemble.start();
 
             // start pulsar service

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/RackAwareTest.java

@@ -0,0 +1,133 @@
+/**
+ * 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.pulsar.broker.service;
+
+import static org.testng.Assert.assertTrue;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.proto.BookieServer;
+import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.apache.bookkeeper.test.PortManager;
+import org.apache.pulsar.common.policies.data.BookieInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+public class RackAwareTest extends BrokerBkEnsemblesTests {
+
+    private static final int NUM_BOOKIES = 6;
+    private final List<BookieServer> bookies = new ArrayList<>();
+
+    public RackAwareTest() {
+        // Start bookies manually
+        super(0);
+    }
+
+    @SuppressWarnings("deprecation")
+    @BeforeClass
+    void setup() throws Exception {
+        super.setup();
+
+        // Start bookies with specific racks
+        for (int i = 0; i < NUM_BOOKIES; i++) {
+            File bkDataDir = Files.createTempDirectory("bk" + Integer.toString(i) + "test").toFile();
+            ServerConfiguration conf = new ServerConfiguration();
+
+            int bookiePort = PortManager.nextFreePort();
+            conf.setBookiePort(bookiePort);
+            conf.setZkServers("127.0.0.1:" + this.bkEnsemble.getZkServer().getClientPort());
+            conf.setJournalDirName(bkDataDir.getPath());
+            conf.setLedgerDirNames(new String[] { bkDataDir.getPath() });
+            conf.setAllowLoopback(true);
+
+            // Use different advertised addresses for each bookie, so we can place them in different
+            // racks.
+            // Eg: 1st bookie will be 10.0.0.1, 2nd 10.0.0.2 and so on
+            String addr = String.format("10.0.0.%d", i + 1);
+            conf.setAdvertisedAddress(addr);
+
+            BookieServer bs = new BookieServer(conf, NullStatsLogger.INSTANCE);
+
+            bs.start();
+            bookies.add(bs);
+            log.info("Local BK[{}] started (port: {}, adddress: {})", i, bookiePort, addr);
+        }
+
+    }
+
+    @AfterClass
+    void shutdown() throws Exception {
+        super.shutdown();
+
+        for (BookieServer bs : bookies) {
+            bs.shutdown();
+        }
+
+        bookies.clear();
+    }
+
+    @Test
+    public void testPlacement() throws Exception {
+        for (int i = 0; i < NUM_BOOKIES; i++) {
+            String bookie = bookies.get(i).getLocalAddress().toString();
+
+            // Place bookie-1 in "rack-1" and the rest in "rack-2"
+            int rackId = i == 0 ? 1 : 2;
+            BookieInfo bi = new BookieInfo("rack-" + rackId, "bookie-" + (i + 1));
+            log.info("setting rack for bookie at {} -- {}", bookie, bi);
+            admin.bookies().updateBookieRackInfo(bookie, "default", bi);
+        }
+
+        // Make sure the racks cache gets updated through the ZK watch
+        Thread.sleep(1000);
+
+        BookKeeper bkc = this.pulsar.getBookKeeperClient();
+
+        // Create few ledgers and verify all of them should have a copy in the first bookie
+        BookieSocketAddress fistBookie = bookies.get(0).getLocalAddress();
+        for (int i = 0; i < 100; i++) {
+            LedgerHandle lh = bkc.createLedger(2, 2, DigestType.DUMMY, new byte[0]);
+            log.info("Ledger: {} -- Ensemble: {}", i, lh.getLedgerMetadata().getEnsembleAt(0));
+            assertTrue(lh.getLedgerMetadata().getEnsembleAt(0).contains(fistBookie),
+                    "first bookie in rack 0 not included in ensemble");
+            lh.close();
+        }
+    }
+
+    public void testCrashBrokerWithoutCursorLedgerLeak() throws Exception {
+        // Ignore test
+    }
+
+    public void testSkipCorruptDataLedger() throws Exception {
+        // Ignore test
+    }
+
+    private static final Logger log = LoggerFactory.getLogger(RackAwareTest.class);
+}

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Bookies.java

@@ -0,0 +1,48 @@
+/**
+ * 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.pulsar.client.admin;
+
+import org.apache.pulsar.common.policies.data.BookieInfo;
+import org.apache.pulsar.common.policies.data.BookiesRackConfiguration;
+
+/**
+ * Admin interface for bookies rack placement management.
+ */
+public interface Bookies {
+
+    /**
+     * Gets the rack placement information for all the bookies in the cluster
+     */
+    BookiesRackConfiguration getBookiesRackInfo() throws PulsarAdminException;
+
+    /**
+     * Gets the rack placement information for a specific bookie in the cluster
+     */
+    BookieInfo getBookieRackInfo(String bookieAddress) throws PulsarAdminException;
+
+    /**
+     * Remove rack placement information for a specific bookie in the cluster
+     */
+    void deleteBookieRackInfo(String bookieAddress) throws PulsarAdminException;
+
+    /**
+     * Updates the rack placement information for a specific bookie in the cluster
+     */
+    void updateBookieRackInfo(String bookieAddress, String group, BookieInfo bookieInfo) throws PulsarAdminException;
+}