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