001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.backup.example;
019
020import static org.junit.Assert.assertEquals;
021import static org.junit.Assert.assertFalse;
022import static org.junit.Assert.assertTrue;
023import static org.mockito.Mockito.mock;
024
025import java.io.IOException;
026import java.util.ArrayList;
027import java.util.List;
028import java.util.concurrent.CompletableFuture;
029import java.util.concurrent.CountDownLatch;
030import org.apache.hadoop.conf.Configuration;
031import org.apache.hadoop.fs.FileStatus;
032import org.apache.hadoop.fs.FileSystem;
033import org.apache.hadoop.fs.Path;
034import org.apache.hadoop.hbase.ChoreService;
035import org.apache.hadoop.hbase.HBaseClassTestRule;
036import org.apache.hadoop.hbase.HBaseTestingUtility;
037import org.apache.hadoop.hbase.HConstants;
038import org.apache.hadoop.hbase.Stoppable;
039import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
040import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
041import org.apache.hadoop.hbase.client.Connection;
042import org.apache.hadoop.hbase.client.ConnectionFactory;
043import org.apache.hadoop.hbase.client.DummyAsyncRegistry;
044import org.apache.hadoop.hbase.client.Put;
045import org.apache.hadoop.hbase.master.cleaner.BaseHFileCleanerDelegate;
046import org.apache.hadoop.hbase.master.cleaner.CleanerChore;
047import org.apache.hadoop.hbase.master.cleaner.HFileCleaner;
048import org.apache.hadoop.hbase.regionserver.CompactedHFilesDischarger;
049import org.apache.hadoop.hbase.regionserver.HRegion;
050import org.apache.hadoop.hbase.regionserver.HStore;
051import org.apache.hadoop.hbase.regionserver.RegionServerServices;
052import org.apache.hadoop.hbase.testclassification.MediumTests;
053import org.apache.hadoop.hbase.testclassification.MiscTests;
054import org.apache.hadoop.hbase.util.Bytes;
055import org.apache.hadoop.hbase.util.FSUtils;
056import org.apache.hadoop.hbase.util.HFileArchiveUtil;
057import org.apache.hadoop.hbase.util.StoppableImplementation;
058import org.apache.hadoop.hbase.zookeeper.ZKUtil;
059import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
060import org.apache.zookeeper.KeeperException;
061import org.junit.After;
062import org.junit.AfterClass;
063import org.junit.BeforeClass;
064import org.junit.ClassRule;
065import org.junit.Test;
066import org.junit.experimental.categories.Category;
067import org.mockito.Mockito;
068import org.mockito.invocation.InvocationOnMock;
069import org.mockito.stubbing.Answer;
070import org.slf4j.Logger;
071import org.slf4j.LoggerFactory;
072
073/**
074 * Spin up a small cluster and check that the hfiles of region are properly long-term archived as
075 * specified via the {@link ZKTableArchiveClient}.
076 */
077@Category({MiscTests.class, MediumTests.class})
078public class TestZooKeeperTableArchiveClient {
079
080  @ClassRule
081  public static final HBaseClassTestRule CLASS_RULE =
082      HBaseClassTestRule.forClass(TestZooKeeperTableArchiveClient.class);
083
084  private static final Logger LOG = LoggerFactory.getLogger(TestZooKeeperTableArchiveClient.class);
085  private static final HBaseTestingUtility UTIL = new HBaseTestingUtility();
086  private static final String STRING_TABLE_NAME = "test";
087  private static final byte[] TEST_FAM = Bytes.toBytes("fam");
088  private static final byte[] TABLE_NAME = Bytes.toBytes(STRING_TABLE_NAME);
089  private static ZKTableArchiveClient archivingClient;
090  private final List<Path> toCleanup = new ArrayList<>();
091  private static Connection CONNECTION;
092  private static RegionServerServices rss;
093
094  public static final class MockRegistry extends DummyAsyncRegistry {
095
096    public MockRegistry(Configuration conf) {
097    }
098
099    @Override
100    public CompletableFuture<String> getClusterId() {
101      return CompletableFuture.completedFuture("clusterId");
102    }
103  }
104
105  /**
106   * Setup the config for the cluster
107   */
108  @BeforeClass
109  public static void setupCluster() throws Exception {
110    setupConf(UTIL.getConfiguration());
111    UTIL.startMiniZKCluster();
112    UTIL.getConfiguration().setClass("hbase.client.registry.impl", MockRegistry.class,
113      DummyAsyncRegistry.class);
114    CONNECTION = ConnectionFactory.createConnection(UTIL.getConfiguration());
115    archivingClient = new ZKTableArchiveClient(UTIL.getConfiguration(), CONNECTION);
116    // make hfile archiving node so we can archive files
117    ZKWatcher watcher = UTIL.getZooKeeperWatcher();
118    String archivingZNode = ZKTableArchiveClient.getArchiveZNode(UTIL.getConfiguration(), watcher);
119    ZKUtil.createWithParents(watcher, archivingZNode);
120    rss = mock(RegionServerServices.class);
121  }
122
123  private static void setupConf(Configuration conf) {
124    // only compact with 3 files
125    conf.setInt("hbase.hstore.compaction.min", 3);
126  }
127
128  @After
129  public void tearDown() throws Exception {
130    try {
131      FileSystem fs = UTIL.getTestFileSystem();
132      // cleanup each of the files/directories registered
133      for (Path file : toCleanup) {
134      // remove the table and archive directories
135        FSUtils.delete(fs, file, true);
136      }
137    } catch (IOException e) {
138      LOG.warn("Failure to delete archive directory", e);
139    } finally {
140      toCleanup.clear();
141    }
142    // make sure that backups are off for all tables
143    archivingClient.disableHFileBackup();
144  }
145
146  @AfterClass
147  public static void cleanupTest() throws Exception {
148    try {
149      CONNECTION.close();
150      UTIL.shutdownMiniZKCluster();
151    } catch (Exception e) {
152      LOG.warn("problem shutting down cluster", e);
153    }
154  }
155
156  /**
157   * Test turning on/off archiving
158   */
159  @Test
160  public void testArchivingEnableDisable() throws Exception {
161    // 1. turn on hfile backups
162    LOG.debug("----Starting archiving");
163    archivingClient.enableHFileBackupAsync(TABLE_NAME);
164    assertTrue("Archving didn't get turned on", archivingClient
165        .getArchivingEnabled(TABLE_NAME));
166
167    // 2. Turn off archiving and make sure its off
168    archivingClient.disableHFileBackup();
169    assertFalse("Archving didn't get turned off.", archivingClient.getArchivingEnabled(TABLE_NAME));
170
171    // 3. Check enable/disable on a single table
172    archivingClient.enableHFileBackupAsync(TABLE_NAME);
173    assertTrue("Archving didn't get turned on", archivingClient
174        .getArchivingEnabled(TABLE_NAME));
175
176    // 4. Turn off archiving and make sure its off
177    archivingClient.disableHFileBackup(TABLE_NAME);
178    assertFalse("Archving didn't get turned off for " + STRING_TABLE_NAME,
179      archivingClient.getArchivingEnabled(TABLE_NAME));
180  }
181
182  @Test
183  public void testArchivingOnSingleTable() throws Exception {
184    createArchiveDirectory();
185    FileSystem fs = UTIL.getTestFileSystem();
186    Path archiveDir = getArchiveDir();
187    Path tableDir = getTableDir(STRING_TABLE_NAME);
188    toCleanup.add(archiveDir);
189    toCleanup.add(tableDir);
190
191    Configuration conf = UTIL.getConfiguration();
192    // setup the delegate
193    Stoppable stop = new StoppableImplementation();
194    CleanerChore.initChorePool(conf);
195    HFileCleaner cleaner = setupAndCreateCleaner(conf, fs, archiveDir, stop);
196    List<BaseHFileCleanerDelegate> cleaners = turnOnArchiving(STRING_TABLE_NAME, cleaner);
197    final LongTermArchivingHFileCleaner delegate = (LongTermArchivingHFileCleaner) cleaners.get(0);
198
199    // create the region
200    ColumnFamilyDescriptor hcd = ColumnFamilyDescriptorBuilder.of(TEST_FAM);
201    HRegion region = UTIL.createTestRegion(STRING_TABLE_NAME, hcd);
202    List<HRegion> regions = new ArrayList<>();
203    regions.add(region);
204    Mockito.doReturn(regions).when(rss).getRegions();
205    final CompactedHFilesDischarger compactionCleaner =
206        new CompactedHFilesDischarger(100, stop, rss, false);
207    loadFlushAndCompact(region, TEST_FAM);
208    compactionCleaner.chore();
209    // get the current hfiles in the archive directory
210    List<Path> files = getAllFiles(fs, archiveDir);
211    if (files == null) {
212      FSUtils.logFileSystemState(fs, UTIL.getDataTestDir(), LOG);
213      throw new RuntimeException("Didn't archive any files!");
214    }
215    CountDownLatch finished = setupCleanerWatching(delegate, cleaners, files.size());
216
217    runCleaner(cleaner, finished, stop);
218
219    // know the cleaner ran, so now check all the files again to make sure they are still there
220    List<Path> archivedFiles = getAllFiles(fs, archiveDir);
221    assertEquals("Archived files changed after running archive cleaner.", files, archivedFiles);
222
223    // but we still have the archive directory
224    assertTrue(fs.exists(HFileArchiveUtil.getArchivePath(UTIL.getConfiguration())));
225  }
226
227  /**
228   * Test archiving/cleaning across multiple tables, where some are retained, and others aren't
229   * @throws Exception on failure
230   */
231  @Test
232  public void testMultipleTables() throws Exception {
233    createArchiveDirectory();
234    String otherTable = "otherTable";
235
236    FileSystem fs = UTIL.getTestFileSystem();
237    Path archiveDir = getArchiveDir();
238    Path tableDir = getTableDir(STRING_TABLE_NAME);
239    Path otherTableDir = getTableDir(otherTable);
240
241    // register cleanup for the created directories
242    toCleanup.add(archiveDir);
243    toCleanup.add(tableDir);
244    toCleanup.add(otherTableDir);
245    Configuration conf = UTIL.getConfiguration();
246    // setup the delegate
247    Stoppable stop = new StoppableImplementation();
248    final ChoreService choreService = new ChoreService("TEST_SERVER_NAME");
249    CleanerChore.initChorePool(conf);
250    HFileCleaner cleaner = setupAndCreateCleaner(conf, fs, archiveDir, stop);
251    List<BaseHFileCleanerDelegate> cleaners = turnOnArchiving(STRING_TABLE_NAME, cleaner);
252    final LongTermArchivingHFileCleaner delegate = (LongTermArchivingHFileCleaner) cleaners.get(0);
253    // create the region
254    ColumnFamilyDescriptor hcd = ColumnFamilyDescriptorBuilder.of(TEST_FAM);
255    HRegion region = UTIL.createTestRegion(STRING_TABLE_NAME, hcd);
256    List<HRegion> regions = new ArrayList<>();
257    regions.add(region);
258    Mockito.doReturn(regions).when(rss).getRegions();
259    final CompactedHFilesDischarger compactionCleaner =
260        new CompactedHFilesDischarger(100, stop, rss, false);
261    loadFlushAndCompact(region, TEST_FAM);
262    compactionCleaner.chore();
263    // create the another table that we don't archive
264    hcd = ColumnFamilyDescriptorBuilder.of(TEST_FAM);
265    HRegion otherRegion = UTIL.createTestRegion(otherTable, hcd);
266    regions = new ArrayList<>();
267    regions.add(otherRegion);
268    Mockito.doReturn(regions).when(rss).getRegions();
269    final CompactedHFilesDischarger compactionCleaner1 = new CompactedHFilesDischarger(100, stop,
270        rss, false);
271    loadFlushAndCompact(otherRegion, TEST_FAM);
272    compactionCleaner1.chore();
273    // get the current hfiles in the archive directory
274    // Should  be archived
275    List<Path> files = getAllFiles(fs, archiveDir);
276    if (files == null) {
277      FSUtils.logFileSystemState(fs, archiveDir, LOG);
278      throw new RuntimeException("Didn't load archive any files!");
279    }
280
281    // make sure we have files from both tables
282    int initialCountForPrimary = 0;
283    int initialCountForOtherTable = 0;
284    for (Path file : files) {
285      String tableName = file.getParent().getParent().getParent().getName();
286      // check to which table this file belongs
287      if (tableName.equals(otherTable)) initialCountForOtherTable++;
288      else if (tableName.equals(STRING_TABLE_NAME)) initialCountForPrimary++;
289    }
290
291    assertTrue("Didn't archive files for:" + STRING_TABLE_NAME, initialCountForPrimary > 0);
292    assertTrue("Didn't archive files for:" + otherTable, initialCountForOtherTable > 0);
293
294    // run the cleaners, checking for each of the directories + files (both should be deleted and
295    // need to be checked) in 'otherTable' and the files (which should be retained) in the 'table'
296    CountDownLatch finished = setupCleanerWatching(delegate, cleaners, files.size() + 3);
297    // run the cleaner
298    choreService.scheduleChore(cleaner);
299    // wait for the cleaner to check all the files
300    finished.await();
301    // stop the cleaner
302    stop.stop("");
303
304    // know the cleaner ran, so now check all the files again to make sure they are still there
305    List<Path> archivedFiles = getAllFiles(fs, archiveDir);
306    int archivedForPrimary = 0;
307    for(Path file: archivedFiles) {
308      String tableName = file.getParent().getParent().getParent().getName();
309      // ensure we don't have files from the non-archived table
310      assertFalse("Have a file from the non-archived table: " + file, tableName.equals(otherTable));
311      if (tableName.equals(STRING_TABLE_NAME)) archivedForPrimary++;
312    }
313
314    assertEquals("Not all archived files for the primary table were retained.", initialCountForPrimary,
315      archivedForPrimary);
316
317    // but we still have the archive directory
318    assertTrue("Archive directory was deleted via archiver", fs.exists(archiveDir));
319  }
320
321
322  private void createArchiveDirectory() throws IOException {
323    //create the archive and test directory
324    FileSystem fs = UTIL.getTestFileSystem();
325    Path archiveDir = getArchiveDir();
326    fs.mkdirs(archiveDir);
327  }
328
329  private Path getArchiveDir() throws IOException {
330    return new Path(UTIL.getDataTestDir(), HConstants.HFILE_ARCHIVE_DIRECTORY);
331  }
332
333  private Path getTableDir(String tableName) throws IOException {
334    Path testDataDir = UTIL.getDataTestDir();
335    FSUtils.setRootDir(UTIL.getConfiguration(), testDataDir);
336    return new Path(testDataDir, tableName);
337  }
338
339  private HFileCleaner setupAndCreateCleaner(Configuration conf, FileSystem fs, Path archiveDir,
340      Stoppable stop) {
341    conf.setStrings(HFileCleaner.MASTER_HFILE_CLEANER_PLUGINS,
342      LongTermArchivingHFileCleaner.class.getCanonicalName());
343    return new HFileCleaner(1000, stop, conf, fs, archiveDir);
344  }
345
346  /**
347   * Start archiving table for given hfile cleaner
348   * @param tableName table to archive
349   * @param cleaner cleaner to check to make sure change propagated
350   * @return underlying {@link LongTermArchivingHFileCleaner} that is managing archiving
351   * @throws IOException on failure
352   * @throws KeeperException on failure
353   */
354  private List<BaseHFileCleanerDelegate> turnOnArchiving(String tableName, HFileCleaner cleaner)
355      throws IOException, KeeperException {
356    // turn on hfile retention
357    LOG.debug("----Starting archiving for table:" + tableName);
358    archivingClient.enableHFileBackupAsync(Bytes.toBytes(tableName));
359    assertTrue("Archving didn't get turned on", archivingClient.getArchivingEnabled(tableName));
360
361    // wait for the archiver to get the notification
362    List<BaseHFileCleanerDelegate> cleaners = cleaner.getDelegatesForTesting();
363    LongTermArchivingHFileCleaner delegate = (LongTermArchivingHFileCleaner) cleaners.get(0);
364    while (!delegate.archiveTracker.keepHFiles(STRING_TABLE_NAME)) {
365      // spin until propagation - should be fast
366    }
367    return cleaners;
368  }
369
370  /**
371   * Spy on the {@link LongTermArchivingHFileCleaner} to ensure we can catch when the cleaner has
372   * seen all the files
373   * @return a {@link CountDownLatch} to wait on that releases when the cleaner has been called at
374   *         least the expected number of times.
375   */
376  private CountDownLatch setupCleanerWatching(LongTermArchivingHFileCleaner cleaner,
377      List<BaseHFileCleanerDelegate> cleaners, final int expected) {
378    // replace the cleaner with one that we can can check
379    BaseHFileCleanerDelegate delegateSpy = Mockito.spy(cleaner);
380    final int[] counter = new int[] { 0 };
381    final CountDownLatch finished = new CountDownLatch(1);
382    Mockito.doAnswer(new Answer<Iterable<FileStatus>>() {
383
384      @Override
385      public Iterable<FileStatus> answer(InvocationOnMock invocation) throws Throwable {
386        counter[0]++;
387        LOG.debug(counter[0] + "/ " + expected + ") Wrapping call to getDeletableFiles for files: "
388            + invocation.getArgument(0));
389
390        @SuppressWarnings("unchecked")
391        Iterable<FileStatus> ret = (Iterable<FileStatus>) invocation.callRealMethod();
392        if (counter[0] >= expected) finished.countDown();
393        return ret;
394      }
395    }).when(delegateSpy).getDeletableFiles(Mockito.anyList());
396    cleaners.set(0, delegateSpy);
397
398    return finished;
399  }
400
401  /**
402   * Get all the files (non-directory entries) in the file system under the passed directory
403   * @param dir directory to investigate
404   * @return all files under the directory
405   */
406  private List<Path> getAllFiles(FileSystem fs, Path dir) throws IOException {
407    FileStatus[] files = FSUtils.listStatus(fs, dir, null);
408    if (files == null) {
409      LOG.warn("No files under:" + dir);
410      return null;
411    }
412
413    List<Path> allFiles = new ArrayList<>();
414    for (FileStatus file : files) {
415      if (file.isDirectory()) {
416        List<Path> subFiles = getAllFiles(fs, file.getPath());
417        if (subFiles != null) allFiles.addAll(subFiles);
418        continue;
419      }
420      allFiles.add(file.getPath());
421    }
422    return allFiles;
423  }
424
425  private void loadFlushAndCompact(HRegion region, byte[] family) throws IOException {
426    // create two hfiles in the region
427    createHFileInRegion(region, family);
428    createHFileInRegion(region, family);
429
430    HStore s = region.getStore(family);
431    int count = s.getStorefilesCount();
432    assertTrue("Don't have the expected store files, wanted >= 2 store files, but was:" + count,
433      count >= 2);
434
435    // compact the two files into one file to get files in the archive
436    LOG.debug("Compacting stores");
437    region.compact(true);
438  }
439
440  /**
441   * Create a new hfile in the passed region
442   * @param region region to operate on
443   * @param columnFamily family for which to add data
444   * @throws IOException
445   */
446  private void createHFileInRegion(HRegion region, byte[] columnFamily) throws IOException {
447    // put one row in the region
448    Put p = new Put(Bytes.toBytes("row"));
449    p.addColumn(columnFamily, Bytes.toBytes("Qual"), Bytes.toBytes("v1"));
450    region.put(p);
451    // flush the region to make a store file
452    region.flush(true);
453  }
454
455  /**
456   * @param cleaner
457   */
458  private void runCleaner(HFileCleaner cleaner, CountDownLatch finished, Stoppable stop)
459      throws InterruptedException {
460    final ChoreService choreService = new ChoreService("CLEANER_SERVER_NAME");
461    // run the cleaner
462    choreService.scheduleChore(cleaner);
463    // wait for the cleaner to check all the files
464    finished.await();
465    // stop the cleaner
466    stop.stop("");
467  }
468}