tests/integration/smoke/src/test/java/org/apache/bookkeeper/tests/integration/TestBookieShellCluster.java

/*
 * 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.tests.integration;

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.Assert.assertTrue;

import com.github.dockerjava.api.DockerClient;
import java.util.concurrent.ExecutionException;
import lombok.extern.slf4j.Slf4j;

import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.api.BookKeeper;
import org.apache.bookkeeper.client.api.LedgerEntries;
import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.client.api.WriteHandle;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.tests.integration.utils.BookKeeperClusterUtils;
import org.apache.bookkeeper.tests.integration.utils.DockerUtils;
import org.jboss.arquillian.junit.Arquillian;
import org.jboss.arquillian.test.api.ArquillianResource;
import org.junit.Assert;
import org.junit.FixMethodOrder;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.MethodSorters;

/**
 * Test cluster related commands in bookie shell.
 */
@Slf4j
@RunWith(Arquillian.class)
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
public class TestBookieShellCluster extends BookieShellTestBase {

    @ArquillianResource
    private DockerClient docker;

    private String currentVersion = System.getProperty("currentVersion");

    private static final byte[] PASSWORD = "foobar".getBytes(UTF_8);

    @Test
    @Override
    public void test000_Setup() throws Exception {
        // First test to run, formats metadata and bookies
        if (BookKeeperClusterUtils.metadataFormatIfNeeded(docker, currentVersion)) {
            BookKeeperClusterUtils.formatAllBookies(docker, currentVersion);
        }
        assertTrue(BookKeeperClusterUtils.startAllBookiesWithVersion(docker, currentVersion));
    }

    @Test
    @Override
    public void test999_Teardown() {
        assertTrue(BookKeeperClusterUtils.stopAllBookies(docker));
    }

    @Override
    protected String runCommandInAnyContainer(String... cmds) throws Exception {
        String bookie = BookKeeperClusterUtils.getAnyBookie();
        return DockerUtils.runCommand(docker, bookie, cmds);
    }

    @Test
    @Override
    public void test001_SimpleTest() throws Exception {
        super.test001_SimpleTest();
    }

    @Test
    @Override
    public void test002_ListROBookies() throws Exception {
        super.test002_ListROBookies();
    }

    @Test
    @Override
    public void test003_ListRWBookies() throws Exception {
        super.test003_ListRWBookies();
    }

    private static long writeNEntries(BookKeeper bk, int n, int numBookies) throws Exception {
        try (WriteHandle writer = bk.newCreateLedgerOp().withEnsembleSize(numBookies)
             .withWriteQuorumSize(numBookies).withAckQuorumSize(numBookies)
             .withPassword(PASSWORD).execute().get()) {
            int i = 0;
            for (; i < n - 1; i++) {
                writer.appendAsync(("entry" + i).getBytes(UTF_8));
            }
            writer.append(("entry" + i).getBytes(UTF_8));

            return writer.getId();
        }
    }

    private static void validateNEntries(BookKeeper bk, long ledgerId, int n) throws Exception {
        try (ReadHandle reader = bk.newOpenLedgerOp()
             .withLedgerId(ledgerId)
             .withPassword(PASSWORD)
             .execute().get();
             LedgerEntries entries = reader.read(0, n - 1)) {
            Assert.assertEquals(reader.getLastAddConfirmed(), n - 1);

            for (int i = 0; i < n; i++) {
                Assert.assertEquals("entry" + i, new String(entries.getEntry(i).getEntryBytes(), UTF_8));
            }
        }
    }

    /**
     * These tests on being able to access cluster internals, so can't be put in test base.
     */
    @Test
    public void test101_RegenerateIndex() throws Exception {
        String zookeeper = String.format("zk+hierarchical://%s/ledgers",
                                         BookKeeperClusterUtils.zookeeperConnectString(docker));
        int numEntries = 100;

        try (BookKeeper bk = BookKeeper.newBuilder(
                     new ClientConfiguration().setMetadataServiceUri(zookeeper)).build()) {
            log.info("Writing entries");
            long ledgerId1 = writeNEntries(bk, numEntries, BookKeeperClusterUtils.allBookies().size());
            long ledgerId2 = writeNEntries(bk, numEntries, BookKeeperClusterUtils.allBookies().size());

            log.info("Validate that we can read back");
            validateNEntries(bk, ledgerId1, numEntries);
            validateNEntries(bk, ledgerId2, numEntries);

            String indexFileName1 = String.format("/opt/bookkeeper/data/ledgers/current/0/%d/%d.idx",
                                                  ledgerId1, ledgerId1);
            String indexFileName2 = String.format("/opt/bookkeeper/data/ledgers/current/0/%d/%d.idx",
                                                  ledgerId2, ledgerId2);

            log.info("Stop bookies to flush, delete the index and start again");
            assertTrue(BookKeeperClusterUtils.stopAllBookies(docker));

            BookKeeperClusterUtils.runOnAllBookies(docker, "rm", indexFileName1, indexFileName2);
            assertTrue(BookKeeperClusterUtils.startAllBookiesWithVersion(docker, currentVersion));

            log.info("Validate that we cannot read back");
            try {
                validateNEntries(bk, ledgerId1, numEntries);
                Assert.fail("Shouldn't have been able to find anything");
            } catch (BKException.BKNoSuchLedgerExistsException e) {
                // expected
            }
            try {
                validateNEntries(bk, ledgerId2, numEntries);
                Assert.fail("Shouldn't have been able to find anything");
            } catch (BKException.BKNoSuchLedgerExistsException e) {
                // expected
            }

            assertTrue(BookKeeperClusterUtils.stopAllBookies(docker));

            log.info("Regenerate the index file");
            BookKeeperClusterUtils.runOnAllBookies(docker,
                    bkScript, "shell", "regenerate-interleaved-storage-index-file",
                    "--ledgerIds", String.format("%d,%d", ledgerId1, ledgerId2),
                    "--password", new String(PASSWORD, UTF_8));
            assertTrue(BookKeeperClusterUtils.startAllBookiesWithVersion(docker, currentVersion));

            log.info("Validate that we can read back, after regeneration");
            validateNEntries(bk, ledgerId1, numEntries);
            validateNEntries(bk, ledgerId2, numEntries);
        }
    }

    @Test
    public void test102_DumpRestoreMetadata() throws Exception {
        String zookeeper = String.format("zk+hierarchical://%s/ledgers",
                                         BookKeeperClusterUtils.zookeeperConnectString(docker));
        int numEntries = 100;

        try (BookKeeper bk = BookKeeper.newBuilder(
                     new ClientConfiguration().setMetadataServiceUri(zookeeper)).build()) {
            log.info("Writing entries");
            long ledgerId = writeNEntries(bk, numEntries, 1);

            log.info("Dumping ledger metadata to file");
            String bookie = BookKeeperClusterUtils.getAnyBookie();
            String dumpFile = String.format("/tmp/ledger-%d-%d", ledgerId, System.nanoTime());
            DockerUtils.runCommand(docker, bookie,
                                   bkScript, "shell", "ledgermetadata",
                                   "--ledgerid", String.valueOf(ledgerId),
                                   "--dumptofile", dumpFile);

            log.info("Delete the ledger metadata");
            bk.newDeleteLedgerOp().withLedgerId(ledgerId).execute().get();

            // hopefully ledger gc doesn't kick in
            log.info("Verify that we cannot open ledger");
            try {
                validateNEntries(bk, ledgerId, numEntries);
                Assert.fail("Shouldn't have been able to find anything");
            } catch (ExecutionException ee) {
                Assert.assertEquals(ee.getCause().getClass(),
                        BKException.BKNoSuchLedgerExistsOnMetadataServerException.class);
            }

            log.info("Restore the ledger metadata");
            DockerUtils.runCommand(docker, bookie,
                                   bkScript, "shell", "ledgermetadata",
                                   "--ledgerid", String.valueOf(ledgerId),
                                   "--restorefromfile", dumpFile);

            log.info("Validate that we can read back, after regeneration");
            validateNEntries(bk, ledgerId, numEntries);
        }
    }

}