Source code

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;
019
020import java.io.IOException;
021import java.util.ArrayList;
022import java.util.Arrays;
023import java.util.Iterator;
024import java.util.List;
025import java.util.Map;
026import java.util.Map.Entry;
027import java.util.Objects;
028import org.apache.hadoop.conf.Configuration;
029import org.apache.hadoop.fs.FileStatus;
030import org.apache.hadoop.fs.FileSystem;
031import org.apache.hadoop.fs.LocatedFileStatus;
032import org.apache.hadoop.fs.Path;
033import org.apache.hadoop.fs.RemoteIterator;
034import org.apache.hadoop.hbase.HBaseConfiguration;
035import org.apache.hadoop.hbase.HBaseTestingUtil;
036import org.apache.hadoop.hbase.HConstants;
037import org.apache.hadoop.hbase.NamespaceDescriptor;
038import org.apache.hadoop.hbase.TableName;
039import org.apache.hadoop.hbase.backup.BackupInfo.BackupPhase;
040import org.apache.hadoop.hbase.backup.BackupInfo.BackupState;
041import org.apache.hadoop.hbase.backup.impl.BackupAdminImpl;
042import org.apache.hadoop.hbase.backup.impl.BackupManager;
043import org.apache.hadoop.hbase.backup.impl.BackupSystemTable;
044import org.apache.hadoop.hbase.backup.impl.FullTableBackupClient;
045import org.apache.hadoop.hbase.backup.impl.IncrementalBackupManager;
046import org.apache.hadoop.hbase.backup.impl.IncrementalTableBackupClient;
047import org.apache.hadoop.hbase.backup.util.BackupUtils;
048import org.apache.hadoop.hbase.client.Admin;
049import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
050import org.apache.hadoop.hbase.client.Connection;
051import org.apache.hadoop.hbase.client.ConnectionFactory;
052import org.apache.hadoop.hbase.client.Durability;
053import org.apache.hadoop.hbase.client.Put;
054import org.apache.hadoop.hbase.client.Table;
055import org.apache.hadoop.hbase.client.TableDescriptor;
056import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
057import org.apache.hadoop.hbase.master.cleaner.LogCleaner;
058import org.apache.hadoop.hbase.master.cleaner.TimeToLiveLogCleaner;
059import org.apache.hadoop.hbase.regionserver.LogRoller;
060import org.apache.hadoop.hbase.security.HadoopSecurityEnabledUserProviderForTesting;
061import org.apache.hadoop.hbase.security.UserProvider;
062import org.apache.hadoop.hbase.security.access.SecureTestUtil;
063import org.apache.hadoop.hbase.snapshot.SnapshotTestingUtils;
064import org.apache.hadoop.hbase.util.Bytes;
065import org.apache.hadoop.hbase.util.CommonFSUtils;
066import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
067import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
068import org.apache.hadoop.hbase.wal.WALFactory;
069import org.junit.AfterClass;
070import org.junit.Before;
071import org.junit.BeforeClass;
072import org.junit.jupiter.api.AfterAll;
073import org.junit.jupiter.api.BeforeAll;
074import org.junit.jupiter.api.BeforeEach;
075import org.slf4j.Logger;
076import org.slf4j.LoggerFactory;
077
078/**
079 * This class is only a base for other integration-level backup tests. Do not add tests here.
080 * TestBackupSmallTests is where tests that don't require bring machines up/down should go All other
081 * tests should have their own classes and extend this one
082 */
083public class TestBackupBase {
084  private static final Logger LOG = LoggerFactory.getLogger(TestBackupBase.class);
085
086  protected static HBaseTestingUtil TEST_UTIL;
087  protected static HBaseTestingUtil TEST_UTIL2;
088  protected static Configuration conf1;
089  protected static Configuration conf2;
090
091  protected static TableName table1 = TableName.valueOf("table1");
092  protected static TableDescriptor table1Desc;
093  protected static TableName table2 = TableName.valueOf("table2");
094  protected static TableName table3 = TableName.valueOf("table3");
095  protected static TableName table4 = TableName.valueOf("table4");
096
097  protected static TableName table1_restore = TableName.valueOf("default:table1");
098  protected static TableName table2_restore = TableName.valueOf("ns2:table2");
099  protected static TableName table3_restore = TableName.valueOf("ns3:table3_restore");
100
101  protected static final int NB_ROWS_IN_BATCH = 99;
102  protected static final byte[] qualName = Bytes.toBytes("q1");
103  protected static final byte[] famName = Bytes.toBytes("f");
104
105  protected static String BACKUP_ROOT_DIR;
106  protected static String BACKUP_REMOTE_ROOT_DIR;
107  protected static String provider = "defaultProvider";
108  protected static boolean secure = false;
109
110  protected static boolean autoRestoreOnFailure;
111  protected static boolean useSecondCluster;
112
113  static class IncrementalTableBackupClientForTest extends IncrementalTableBackupClient {
114    public IncrementalTableBackupClientForTest() {
115    }
116
117    public IncrementalTableBackupClientForTest(Connection conn, String backupId,
118      BackupRequest request) throws IOException {
119      super(conn, backupId, request);
120    }
121
122    @BeforeEach
123    @Before
124    public void ensurePreviousBackupTestsAreCleanedUp() throws Exception {
125      // Every operation here may not be necessary for any given test,
126      // some often being no-ops. the goal is to help ensure atomicity
127      // of that tests that implement TestBackupBase
128      try (BackupAdmin backupAdmin = getBackupAdmin()) {
129        backupManager.finishBackupSession();
130        backupAdmin.listBackupSets().forEach(backupSet -> {
131          try {
132            backupAdmin.deleteBackupSet(backupSet.getName());
133          } catch (IOException ignored) {
134          }
135        });
136      } catch (Exception ignored) {
137      }
138      Arrays.stream(TEST_UTIL.getAdmin().listTableNames())
139        .filter(tableName -> !tableName.isSystemTable()).forEach(tableName -> {
140          try {
141            TEST_UTIL.truncateTable(tableName);
142          } catch (IOException ignored) {
143          }
144        });
145      TEST_UTIL.getMiniHBaseCluster().getRegionServerThreads().forEach(rst -> {
146        try {
147          LogRoller walRoller = rst.getRegionServer().getWalRoller();
148          walRoller.requestRollAll();
149          walRoller.waitUntilWalRollFinished();
150        } catch (Exception ignored) {
151        }
152      });
153    }
154
155    @Override
156    public void execute() throws IOException {
157      // case INCREMENTAL_COPY:
158      try {
159        // case PREPARE_INCREMENTAL:
160        failStageIf(Stage.stage_0);
161        beginBackup(backupManager, backupInfo);
162
163        failStageIf(Stage.stage_1);
164        backupInfo.setPhase(BackupPhase.PREPARE_INCREMENTAL);
165        LOG.debug("For incremental backup, current table set is "
166          + backupManager.getIncrementalBackupTableSet());
167        newTimestamps = ((IncrementalBackupManager) backupManager).getIncrBackupLogFileMap();
168        // copy out the table and region info files for each table
169        BackupUtils.copyTableRegionInfo(conn, backupInfo, conf);
170        // convert WAL to HFiles and copy them to .tmp under BACKUP_ROOT
171        convertWALsToHFiles();
172        incrementalCopyHFiles(new String[] { getBulkOutputDir().toString() },
173          backupInfo.getBackupRootDir());
174        failStageIf(Stage.stage_2);
175
176        // case INCR_BACKUP_COMPLETE:
177        // set overall backup status: complete. Here we make sure to complete the backup.
178        // After this checkpoint, even if entering cancel process, will let the backup finished
179        // Set the previousTimestampMap which is before this current log roll to the manifest.
180        Map<TableName, Map<String, Long>> previousTimestampMap =
181          backupManager.readLogTimestampMap();
182        backupInfo.setIncrTimestampMap(previousTimestampMap);
183
184        // The table list in backupInfo is good for both full backup and incremental backup.
185        // For incremental backup, it contains the incremental backup table set.
186        backupManager.writeRegionServerLogTimestamp(backupInfo.getTables(), newTimestamps);
187        failStageIf(Stage.stage_3);
188
189        Map<TableName, Map<String, Long>> newTableSetTimestampMap =
190          backupManager.readLogTimestampMap();
191
192        Long newStartCode =
193          BackupUtils.getMinValue(BackupUtils.getRSLogTimestampMins(newTableSetTimestampMap));
194        backupManager.writeBackupStartCode(newStartCode);
195
196        handleBulkLoad(backupInfo.getTableNames());
197        failStageIf(Stage.stage_4);
198
199        // backup complete
200        completeBackup(conn, backupInfo, BackupType.INCREMENTAL, conf);
201
202      } catch (Exception e) {
203        failBackup(conn, backupInfo, backupManager, e, "Unexpected Exception : ",
204          BackupType.INCREMENTAL, conf);
205        throw new IOException(e);
206      }
207    }
208  }
209
210  static class FullTableBackupClientForTest extends FullTableBackupClient {
211    public FullTableBackupClientForTest() {
212    }
213
214    public FullTableBackupClientForTest(Connection conn, String backupId, BackupRequest request)
215      throws IOException {
216      super(conn, backupId, request);
217    }
218
219    @Override
220    public void execute() throws IOException {
221      // Get the stage ID to fail on
222      try (Admin admin = conn.getAdmin()) {
223        // Begin BACKUP
224        beginBackup(backupManager, backupInfo);
225        failStageIf(Stage.stage_0);
226        String savedStartCode;
227        boolean firstBackup;
228        // do snapshot for full table backup
229        savedStartCode = backupManager.readBackupStartCode();
230        firstBackup = savedStartCode == null || Long.parseLong(savedStartCode) == 0L;
231        if (firstBackup) {
232          // This is our first backup. Let's put some marker to system table so that we can hold the
233          // logs while we do the backup.
234          backupManager.writeBackupStartCode(0L);
235        }
236        failStageIf(Stage.stage_1);
237        // We roll log here before we do the snapshot. It is possible there is duplicate data
238        // in the log that is already in the snapshot. But if we do it after the snapshot, we
239        // could have data loss.
240        // A better approach is to do the roll log on each RS in the same global procedure as
241        // the snapshot.
242        LOG.info("Execute roll log procedure for full backup ...");
243
244        BackupUtils.logRoll(conn, backupInfo.getBackupRootDir(), conf);
245        failStageIf(Stage.stage_2);
246        newTimestamps = backupManager.readRegionServerLastLogRollResult();
247
248        // SNAPSHOT_TABLES:
249        backupInfo.setPhase(BackupPhase.SNAPSHOT);
250        for (TableName tableName : tableList) {
251          String snapshotName = "snapshot_" + Long.toString(EnvironmentEdgeManager.currentTime())
252            + "_" + tableName.getNamespaceAsString() + "_" + tableName.getQualifierAsString();
253
254          snapshotTable(admin, tableName, snapshotName);
255          backupInfo.setSnapshotName(tableName, snapshotName);
256        }
257        failStageIf(Stage.stage_3);
258        // SNAPSHOT_COPY:
259        // do snapshot copy
260        LOG.debug("snapshot copy for " + backupId);
261        snapshotCopy(backupInfo);
262        // Updates incremental backup table set
263        backupManager.addIncrementalBackupTableSet(backupInfo.getTables());
264
265        // BACKUP_COMPLETE:
266        // set overall backup status: complete. Here we make sure to complete the backup.
267        // After this checkpoint, even if entering cancel process, will let the backup finished
268        backupInfo.setState(BackupState.COMPLETE);
269        // The table list in backupInfo is good for both full backup and incremental backup.
270        // For incremental backup, it contains the incremental backup table set.
271        backupManager.writeRegionServerLogTimestamp(backupInfo.getTables(), newTimestamps);
272
273        Map<TableName, Map<String, Long>> newTableSetTimestampMap =
274          backupManager.readLogTimestampMap();
275
276        Long newStartCode =
277          BackupUtils.getMinValue(BackupUtils.getRSLogTimestampMins(newTableSetTimestampMap));
278        backupManager.writeBackupStartCode(newStartCode);
279        failStageIf(Stage.stage_4);
280        // backup complete
281        completeBackup(conn, backupInfo, BackupType.FULL, conf);
282
283      } catch (Exception e) {
284
285        if (autoRestoreOnFailure) {
286          failBackup(conn, backupInfo, backupManager, e, "Unexpected BackupException : ",
287            BackupType.FULL, conf);
288        }
289        throw new IOException(e);
290      }
291    }
292  }
293
294  public static void setUpHelper() throws Exception {
295    BACKUP_ROOT_DIR = Path.SEPARATOR + "backupUT";
296    BACKUP_REMOTE_ROOT_DIR = Path.SEPARATOR + "backupUT";
297
298    if (secure) {
299      // set the always on security provider
300      UserProvider.setUserProviderForTesting(TEST_UTIL.getConfiguration(),
301        HadoopSecurityEnabledUserProviderForTesting.class);
302      // setup configuration
303      SecureTestUtil.enableSecurity(TEST_UTIL.getConfiguration());
304    }
305    conf1.setBoolean(BackupRestoreConstants.BACKUP_ENABLE_KEY, true);
306    BackupManager.decorateMasterConfiguration(conf1);
307    BackupManager.decorateRegionServerConfiguration(conf1);
308    conf1.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/1");
309    // Set TTL for old WALs to 1 sec to enforce fast cleaning of an archived
310    // WAL files
311    conf1.setLong(TimeToLiveLogCleaner.TTL_CONF_KEY, 1000);
312    conf1.setLong(LogCleaner.OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC, 1000);
313
314    // Set MultiWAL (with 2 default WAL files per RS)
315    conf1.set(WALFactory.WAL_PROVIDER, provider);
316    TEST_UTIL.startMiniCluster();
317    conf1 = TEST_UTIL.getConfiguration();
318    TEST_UTIL.startMiniMapReduceCluster();
319
320    if (useSecondCluster) {
321      conf2 = HBaseConfiguration.create(conf1);
322      conf2.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/2");
323      TEST_UTIL2 = new HBaseTestingUtil(conf2);
324      TEST_UTIL2.setZkCluster(TEST_UTIL.getZkCluster());
325      TEST_UTIL2.startMiniDFSCluster(3);
326      String root2 = TEST_UTIL2.getConfiguration().get("fs.defaultFS");
327      Path p = new Path(new Path(root2), "/tmp/wal");
328      CommonFSUtils.setWALRootDir(TEST_UTIL2.getConfiguration(), p);
329      TEST_UTIL2.startMiniCluster();
330    }
331
332    BACKUP_ROOT_DIR =
333      new Path(new Path(TEST_UTIL.getConfiguration().get("fs.defaultFS")), BACKUP_ROOT_DIR)
334        .toString();
335    LOG.info("ROOTDIR " + BACKUP_ROOT_DIR);
336    if (useSecondCluster) {
337      BACKUP_REMOTE_ROOT_DIR = new Path(
338        new Path(TEST_UTIL2.getConfiguration().get("fs.defaultFS")) + BACKUP_REMOTE_ROOT_DIR)
339        .toString();
340      LOG.info("REMOTE ROOTDIR " + BACKUP_REMOTE_ROOT_DIR);
341    }
342    createTables();
343    populateFromMasterConfig(TEST_UTIL.getHBaseCluster().getMaster().getConfiguration(), conf1);
344  }
345
346  /**
347   * Setup Cluster with appropriate configurations before running tests.
348   * @throws Exception if starting the mini cluster or setting up the tables fails
349   */
350  @BeforeAll
351  @BeforeClass
352  public static void setUp() throws Exception {
353    TEST_UTIL = new HBaseTestingUtil();
354    conf1 = TEST_UTIL.getConfiguration();
355    autoRestoreOnFailure = true;
356    useSecondCluster = false;
357    setUpHelper();
358  }
359
360  private static void populateFromMasterConfig(Configuration masterConf, Configuration conf) {
361    Iterator<Entry<String, String>> it = masterConf.iterator();
362    while (it.hasNext()) {
363      Entry<String, String> e = it.next();
364      conf.set(e.getKey(), e.getValue());
365    }
366  }
367
368  @AfterAll
369  @AfterClass
370  public static void tearDown() throws Exception {
371    try {
372      SnapshotTestingUtils.deleteAllSnapshots(TEST_UTIL.getAdmin());
373    } catch (Exception e) {
374    }
375    SnapshotTestingUtils.deleteArchiveDirectory(TEST_UTIL);
376    if (useSecondCluster) {
377      TEST_UTIL2.shutdownMiniCluster();
378    }
379    TEST_UTIL.shutdownMiniCluster();
380    TEST_UTIL.shutdownMiniMapReduceCluster();
381    autoRestoreOnFailure = true;
382    useSecondCluster = false;
383  }
384
385  Table insertIntoTable(Connection conn, TableName table, byte[] family, int id, int numRows)
386    throws IOException {
387    Table t = conn.getTable(table);
388    Put p1;
389    for (int i = 0; i < numRows; i++) {
390      p1 = new Put(Bytes.toBytes("row-" + table + "-" + id + "-" + i));
391      p1.addColumn(family, qualName, Bytes.toBytes("val" + i));
392      t.put(p1);
393    }
394    return t;
395  }
396
397  protected BackupRequest createBackupRequest(BackupType type, List<TableName> tables,
398    String path) {
399    return createBackupRequest(type, tables, path, false);
400  }
401
402  protected BackupRequest createBackupRequest(BackupType type, List<TableName> tables, String path,
403    boolean noChecksumVerify) {
404    BackupRequest.Builder builder = new BackupRequest.Builder();
405    BackupRequest request = builder.withBackupType(type).withTableList(tables)
406      .withTargetRootDir(path).withNoChecksumVerify(noChecksumVerify).build();
407    return request;
408  }
409
410  protected String backupTables(BackupType type, List<TableName> tables, String path)
411    throws IOException {
412    Connection conn = null;
413    BackupAdmin badmin = null;
414    String backupId;
415    try {
416      conn = ConnectionFactory.createConnection(conf1);
417      badmin = new BackupAdminImpl(conn);
418      BackupRequest request = createBackupRequest(type, new ArrayList<>(tables), path);
419      backupId = badmin.backupTables(request);
420    } finally {
421      if (badmin != null) {
422        badmin.close();
423      }
424      if (conn != null) {
425        conn.close();
426      }
427    }
428    return backupId;
429  }
430
431  protected String fullTableBackup(List<TableName> tables) throws IOException {
432    return backupTables(BackupType.FULL, tables, BACKUP_ROOT_DIR);
433  }
434
435  protected String incrementalTableBackup(List<TableName> tables) throws IOException {
436    return backupTables(BackupType.INCREMENTAL, tables, BACKUP_ROOT_DIR);
437  }
438
439  protected static void loadTable(Table table) throws Exception {
440    Put p; // 100 + 1 row to t1_syncup
441    for (int i = 0; i < NB_ROWS_IN_BATCH; i++) {
442      p = new Put(Bytes.toBytes("row" + i));
443      p.setDurability(Durability.SKIP_WAL);
444      p.addColumn(famName, qualName, Bytes.toBytes("val" + i));
445      table.put(p);
446    }
447  }
448
449  protected static void createTables() throws Exception {
450    long tid = EnvironmentEdgeManager.currentTime();
451    table1 = TableName.valueOf("test-" + tid);
452    Admin ha = TEST_UTIL.getAdmin();
453
454    // Create namespaces
455    ha.createNamespace(NamespaceDescriptor.create("ns1").build());
456    ha.createNamespace(NamespaceDescriptor.create("ns2").build());
457    ha.createNamespace(NamespaceDescriptor.create("ns3").build());
458    ha.createNamespace(NamespaceDescriptor.create("ns4").build());
459
460    TableDescriptor desc = TableDescriptorBuilder.newBuilder(table1)
461      .setColumnFamily(ColumnFamilyDescriptorBuilder.of(famName)).build();
462    ha.createTable(desc);
463    table1Desc = desc;
464    Connection conn = ConnectionFactory.createConnection(conf1);
465    Table table = conn.getTable(table1);
466    loadTable(table);
467    table.close();
468    table2 = TableName.valueOf("ns2:test-" + tid + 1);
469    desc = TableDescriptorBuilder.newBuilder(table2)
470      .setColumnFamily(ColumnFamilyDescriptorBuilder.of(famName)).build();
471    ha.createTable(desc);
472    table = conn.getTable(table2);
473    loadTable(table);
474    table.close();
475    table3 = TableName.valueOf("ns3:test-" + tid + 2);
476    table = TEST_UTIL.createTable(table3, famName);
477    table.close();
478    table4 = TableName.valueOf("ns4:test-" + tid + 3);
479    table = TEST_UTIL.createTable(table4, famName);
480    table.close();
481    ha.close();
482    conn.close();
483  }
484
485  protected boolean checkSucceeded(String backupId) throws IOException {
486    BackupInfo status = getBackupInfo(backupId);
487
488    if (status == null) {
489      return false;
490    }
491
492    return status.getState() == BackupState.COMPLETE;
493  }
494
495  protected boolean checkFailed(String backupId) throws IOException {
496    BackupInfo status = getBackupInfo(backupId);
497
498    if (status == null) {
499      return false;
500    }
501
502    return status.getState() == BackupState.FAILED;
503  }
504
505  private BackupInfo getBackupInfo(String backupId) throws IOException {
506    try (BackupSystemTable table = new BackupSystemTable(TEST_UTIL.getConnection())) {
507      BackupInfo status = table.readBackupInfo(backupId);
508      return status;
509    }
510  }
511
512  protected static BackupAdmin getBackupAdmin() throws IOException {
513    return new BackupAdminImpl(TEST_UTIL.getConnection());
514  }
515
516  /**
517   * Helper method
518   */
519  protected List<TableName> toList(String... args) {
520    List<TableName> ret = new ArrayList<>();
521    for (int i = 0; i < args.length; i++) {
522      ret.add(TableName.valueOf(args[i]));
523    }
524    return ret;
525  }
526
527  protected List<FileStatus> getListOfWALFiles(Configuration c) throws IOException {
528    Path logRoot = new Path(CommonFSUtils.getWALRootDir(c), HConstants.HREGION_LOGDIR_NAME);
529    FileSystem fs = logRoot.getFileSystem(c);
530    RemoteIterator<LocatedFileStatus> it = fs.listFiles(logRoot, true);
531    List<FileStatus> logFiles = new ArrayList<FileStatus>();
532    while (it.hasNext()) {
533      LocatedFileStatus lfs = it.next();
534      if (lfs.isFile() && !AbstractFSWALProvider.isMetaFile(lfs.getPath())) {
535        logFiles.add(lfs);
536        LOG.info(Objects.toString(lfs));
537      }
538    }
539    return logFiles;
540  }
541
542  protected void dumpBackupDir() throws IOException {
543    // Dump Backup Dir
544    FileSystem fs = FileSystem.get(conf1);
545    RemoteIterator<LocatedFileStatus> it = fs.listFiles(new Path(BACKUP_ROOT_DIR), true);
546    while (it.hasNext()) {
547      LOG.debug(Objects.toString(it.next().getPath()));
548    }
549  }
550}