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