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.impl;
019
020import java.io.IOException;
021import java.util.ArrayList;
022import java.util.Collections;
023import java.util.HashSet;
024import java.util.List;
025import java.util.Map;
026import java.util.Objects;
027import java.util.Set;
028import org.apache.hadoop.conf.Configuration;
029import org.apache.hadoop.fs.FileStatus;
030import org.apache.hadoop.fs.FileSystem;
031import org.apache.hadoop.fs.Path;
032import org.apache.hadoop.hbase.TableName;
033import org.apache.hadoop.hbase.backup.BackupInfo;
034import org.apache.hadoop.hbase.backup.BackupInfo.BackupPhase;
035import org.apache.hadoop.hbase.backup.BackupInfo.BackupState;
036import org.apache.hadoop.hbase.backup.BackupRequest;
037import org.apache.hadoop.hbase.backup.BackupRestoreConstants;
038import org.apache.hadoop.hbase.backup.BackupType;
039import org.apache.hadoop.hbase.backup.HBackupFileSystem;
040import org.apache.hadoop.hbase.backup.impl.BackupManifest.BackupImage;
041import org.apache.hadoop.hbase.client.Admin;
042import org.apache.hadoop.hbase.client.Connection;
043import org.apache.hadoop.hbase.util.CommonFSUtils;
044import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
045import org.apache.yetus.audience.InterfaceAudience;
046import org.slf4j.Logger;
047import org.slf4j.LoggerFactory;
048
049/**
050 * Base class for backup operation. Concrete implementation for full and incremental backup are
051 * delegated to corresponding sub-classes: {@link FullTableBackupClient} and
052 * {@link IncrementalTableBackupClient}
053 */
054@InterfaceAudience.Private
055public abstract class TableBackupClient {
056
057  public static final String BACKUP_CLIENT_IMPL_CLASS = "backup.client.impl.class";
058
059  public static final String BACKUP_TEST_MODE_STAGE = "backup.test.mode.stage";
060
061  private static final Logger LOG = LoggerFactory.getLogger(TableBackupClient.class);
062
063  protected Configuration conf;
064  protected Connection conn;
065  protected String backupId;
066  protected List<TableName> tableList;
067  protected Map<String, Long> newTimestamps = null;
068
069  protected BackupManager backupManager;
070  protected BackupInfo backupInfo;
071  protected FileSystem fs;
072
073  public TableBackupClient() {
074  }
075
076  public TableBackupClient(final Connection conn, final String backupId, BackupRequest request)
077    throws IOException {
078    init(conn, backupId, request);
079  }
080
081  public void init(final Connection conn, final String backupId, BackupRequest request)
082    throws IOException {
083    if (request.getBackupType() == BackupType.FULL) {
084      backupManager = new BackupManager(conn, conn.getConfiguration());
085    } else {
086      backupManager = new IncrementalBackupManager(conn, conn.getConfiguration());
087    }
088    this.backupId = backupId;
089    this.tableList = request.getTableList();
090    this.conn = conn;
091    this.conf = conn.getConfiguration();
092    this.fs = CommonFSUtils.getCurrentFileSystem(conf);
093    backupInfo = backupManager.createBackupInfo(backupId, request.getBackupType(), tableList,
094      request.getTargetRootDir(), request.getTotalTasks(), request.getBandwidth());
095    if (tableList == null || tableList.isEmpty()) {
096      this.tableList = new ArrayList<>(backupInfo.getTables());
097    }
098    // Start new session
099    backupManager.startBackupSession();
100  }
101
102  /**
103   * Begin the overall backup.
104   * @param backupInfo backup info
105   * @throws IOException exception
106   */
107  protected void beginBackup(BackupManager backupManager, BackupInfo backupInfo)
108    throws IOException {
109
110    BackupSystemTable.snapshot(conn);
111    backupManager.setBackupInfo(backupInfo);
112    // set the start timestamp of the overall backup
113    long startTs = EnvironmentEdgeManager.currentTime();
114    backupInfo.setStartTs(startTs);
115    // set overall backup status: ongoing
116    backupInfo.setState(BackupState.RUNNING);
117    backupInfo.setPhase(BackupPhase.REQUEST);
118    LOG.info("Backup " + backupInfo.getBackupId() + " started at " + startTs + ".");
119
120    backupManager.updateBackupInfo(backupInfo);
121    if (LOG.isDebugEnabled()) {
122      LOG.debug("Backup session " + backupInfo.getBackupId() + " has been started.");
123    }
124  }
125
126  protected String getMessage(Exception e) {
127    String msg = e.getMessage();
128    if (msg == null || msg.equals("")) {
129      msg = e.getClass().getName();
130    }
131    return msg;
132  }
133
134  /**
135   * Delete HBase snapshot for backup.
136   * @param backupInfo backup info
137   * @throws IOException exception
138   */
139  protected static void deleteSnapshots(final Connection conn, BackupInfo backupInfo,
140    Configuration conf) throws IOException {
141    LOG.debug("Trying to delete snapshot for full backup.");
142    for (String snapshotName : backupInfo.getSnapshotNames()) {
143      if (snapshotName == null) {
144        continue;
145      }
146      LOG.debug("Trying to delete snapshot: " + snapshotName);
147
148      try (Admin admin = conn.getAdmin()) {
149        admin.deleteSnapshot(snapshotName);
150      }
151      LOG.debug("Deleting the snapshot " + snapshotName + " for backup " + backupInfo.getBackupId()
152        + " succeeded.");
153    }
154  }
155
156  /**
157   * Clean up directories with prefix "exportSnapshot-", which are generated when exporting
158   * snapshots.
159   * @throws IOException exception
160   */
161  protected static void cleanupExportSnapshotLog(Configuration conf) throws IOException {
162    FileSystem fs = CommonFSUtils.getCurrentFileSystem(conf);
163    Path stagingDir = new Path(
164      conf.get(BackupRestoreConstants.CONF_STAGING_ROOT, fs.getWorkingDirectory().toString()));
165    FileStatus[] files = CommonFSUtils.listStatus(fs, stagingDir);
166    if (files == null) {
167      return;
168    }
169    for (FileStatus file : files) {
170      if (file.getPath().getName().startsWith("exportSnapshot-")) {
171        LOG.debug("Delete log files of exporting snapshot: " + file.getPath().getName());
172        if (CommonFSUtils.delete(fs, file.getPath(), true) == false) {
173          LOG.warn("Can not delete " + file.getPath());
174        }
175      }
176    }
177  }
178
179  /**
180   * Clean up the uncompleted data at target directory if the ongoing backup has already entered the
181   * copy phase.
182   */
183  protected static void cleanupTargetDir(BackupInfo backupInfo, Configuration conf) {
184    try {
185      // clean up the uncompleted data at target directory if the ongoing backup has already entered
186      // the copy phase
187      LOG.debug("Trying to cleanup up target dir. Current backup phase: " + backupInfo.getPhase());
188      if (
189        backupInfo.getPhase().equals(BackupPhase.SNAPSHOTCOPY)
190          || backupInfo.getPhase().equals(BackupPhase.INCREMENTAL_COPY)
191          || backupInfo.getPhase().equals(BackupPhase.STORE_MANIFEST)
192      ) {
193        FileSystem outputFs = FileSystem.get(new Path(backupInfo.getBackupRootDir()).toUri(), conf);
194
195        // now treat one backup as a transaction, clean up data that has been partially copied at
196        // table level
197        for (TableName table : backupInfo.getTables()) {
198          Path targetDirPath = new Path(HBackupFileSystem
199            .getTableBackupDir(backupInfo.getBackupRootDir(), backupInfo.getBackupId(), table));
200          if (outputFs.delete(targetDirPath, true)) {
201            LOG.debug(
202              "Cleaning up uncompleted backup data at " + targetDirPath.toString() + " done.");
203          } else {
204            LOG.debug("No data has been copied to " + targetDirPath.toString() + ".");
205          }
206
207          Path tableDir = targetDirPath.getParent();
208          FileStatus[] backups = CommonFSUtils.listStatus(outputFs, tableDir);
209          if (backups == null || backups.length == 0) {
210            outputFs.delete(tableDir, true);
211            LOG.debug(tableDir.toString() + " is empty, remove it.");
212          }
213        }
214      }
215
216    } catch (IOException e1) {
217      LOG.error("Cleaning up uncompleted backup data of " + backupInfo.getBackupId() + " at "
218        + backupInfo.getBackupRootDir() + " failed due to " + e1.getMessage() + ".");
219    }
220  }
221
222  /**
223   * Fail the overall backup.
224   * @param backupInfo backup info
225   * @param e          exception
226   * @throws IOException exception
227   */
228  protected void failBackup(Connection conn, BackupInfo backupInfo, BackupManager backupManager,
229    Exception e, String msg, BackupType type, Configuration conf) throws IOException {
230    try {
231      LOG.error(msg + getMessage(e), e);
232      // If this is a cancel exception, then we've already cleaned.
233      // set the failure timestamp of the overall backup
234      backupInfo.setCompleteTs(EnvironmentEdgeManager.currentTime());
235      // set failure message
236      backupInfo.setFailedMsg(e.getMessage());
237      // set overall backup status: failed
238      backupInfo.setState(BackupState.FAILED);
239      // compose the backup failed data
240      String backupFailedData = "BackupId=" + backupInfo.getBackupId() + ",startts="
241        + backupInfo.getStartTs() + ",failedts=" + backupInfo.getCompleteTs() + ",failedphase="
242        + backupInfo.getPhase() + ",failedmessage=" + backupInfo.getFailedMsg();
243      LOG.error(backupFailedData);
244      cleanupAndRestoreBackupSystem(conn, backupInfo, conf);
245      // If backup session is updated to FAILED state - means we
246      // processed recovery already.
247      backupManager.updateBackupInfo(backupInfo);
248      backupManager.finishBackupSession();
249      LOG.error("Backup " + backupInfo.getBackupId() + " failed.");
250    } catch (IOException ee) {
251      LOG.error("Please run backup repair tool manually to restore backup system integrity");
252      throw ee;
253    }
254  }
255
256  public static void cleanupAndRestoreBackupSystem(Connection conn, BackupInfo backupInfo,
257    Configuration conf) throws IOException {
258    BackupType type = backupInfo.getType();
259    // if full backup, then delete HBase snapshots if there already are snapshots taken
260    // and also clean up export snapshot log files if exist
261    if (type == BackupType.FULL) {
262      deleteSnapshots(conn, backupInfo, conf);
263      cleanupExportSnapshotLog(conf);
264    }
265    BackupSystemTable.restoreFromSnapshot(conn);
266    BackupSystemTable.deleteSnapshot(conn);
267    // clean up the uncompleted data at target directory if the ongoing backup has already entered
268    // the copy phase
269    // For incremental backup, DistCp logs will be cleaned with the targetDir.
270    cleanupTargetDir(backupInfo, conf);
271  }
272
273  /**
274   * Creates a manifest based on the provided info, and store it in the backup-specific directory.
275   * @param backupInfo The current backup info
276   * @throws IOException exception
277   */
278  protected void addManifest(BackupInfo backupInfo, BackupType type, Configuration conf)
279    throws IOException {
280    // set the overall backup phase : store manifest
281    backupInfo.setPhase(BackupPhase.STORE_MANIFEST);
282
283    BackupManifest manifest = new BackupManifest(backupInfo);
284    if (type == BackupType.INCREMENTAL) {
285      // set the table region server start and end timestamps for incremental backup
286      manifest.setIncrTimestampMap(backupInfo.getIncrTimestampMap());
287    }
288    List<BackupImage> ancestors = getAncestors(backupInfo);
289    for (BackupImage image : ancestors) {
290      manifest.addDependentImage(image);
291    }
292    manifest.store(conf);
293  }
294
295  /**
296   * Gets the direct ancestors of the currently being created backup.
297   * @param backupInfo The backup info for the backup being created
298   */
299  protected List<BackupImage> getAncestors(BackupInfo backupInfo) throws IOException {
300    LOG.debug("Getting the direct ancestors of the current backup {}", backupInfo.getBackupId());
301
302    // Full backups do not have ancestors
303    if (backupInfo.getType() == BackupType.FULL) {
304      LOG.debug("Current backup is a full backup, no direct ancestor for it.");
305      return Collections.emptyList();
306    }
307
308    List<BackupImage> ancestors = new ArrayList<>();
309    Set<TableName> tablesToCover = new HashSet<>(backupInfo.getTables());
310
311    // Go over the backup history list from newest to oldest
312    List<BackupInfo> allHistoryList = backupManager.getBackupHistory(true);
313    for (BackupInfo backup : allHistoryList) {
314      // If the image has a different rootDir, it cannot be an ancestor.
315      if (!Objects.equals(backup.getBackupRootDir(), backupInfo.getBackupRootDir())) {
316        continue;
317      }
318
319      BackupImage.Builder builder = BackupImage.newBuilder();
320      BackupImage image = builder.withBackupId(backup.getBackupId()).withType(backup.getType())
321        .withRootDir(backup.getBackupRootDir()).withTableList(backup.getTableNames())
322        .withStartTime(backup.getStartTs()).withCompleteTime(backup.getCompleteTs()).build();
323
324      // The ancestors consist of the most recent FULL backups that cover the list of tables
325      // required in the new backup and all INCREMENTAL backups that came after one of those FULL
326      // backups.
327      if (backup.getType().equals(BackupType.INCREMENTAL)) {
328        ancestors.add(image);
329        LOG.debug("Dependent incremental backup image: {BackupID={}}", image.getBackupId());
330      } else {
331        if (tablesToCover.removeAll(new HashSet<>(image.getTableNames()))) {
332          ancestors.add(image);
333          LOG.debug("Dependent full backup image: {BackupID={}}", image.getBackupId());
334
335          if (tablesToCover.isEmpty()) {
336            LOG.debug("Got {} ancestors for the current backup.", ancestors.size());
337            return Collections.unmodifiableList(ancestors);
338          }
339        }
340      }
341    }
342
343    throw new IllegalStateException(
344      "Unable to find full backup that contains tables: " + tablesToCover);
345  }
346
347  /**
348   * Get backup request meta data dir as string.
349   * @param backupInfo backup info
350   * @return meta data dir
351   */
352  protected String obtainBackupMetaDataStr(BackupInfo backupInfo) {
353    StringBuilder sb = new StringBuilder();
354    sb.append("type=" + backupInfo.getType() + ",tablelist=");
355    for (TableName table : backupInfo.getTables()) {
356      sb.append(table + ";");
357    }
358    if (sb.lastIndexOf(";") > 0) {
359      sb.delete(sb.lastIndexOf(";"), sb.lastIndexOf(";") + 1);
360    }
361    sb.append(",targetRootDir=" + backupInfo.getBackupRootDir());
362
363    return sb.toString();
364  }
365
366  /**
367   * Complete the overall backup.
368   * @param backupInfo backup info
369   * @throws IOException exception
370   */
371  protected void completeBackup(final Connection conn, BackupInfo backupInfo, BackupType type,
372    Configuration conf) throws IOException {
373    // set the complete timestamp of the overall backup
374    backupInfo.setCompleteTs(EnvironmentEdgeManager.currentTime());
375    // set overall backup status: complete
376    backupInfo.setState(BackupState.COMPLETE);
377    backupInfo.setProgress(100);
378    // add and store the manifest for the backup
379    addManifest(backupInfo, type, conf);
380
381    // compose the backup complete data
382    String backupCompleteData =
383      obtainBackupMetaDataStr(backupInfo) + ",startts=" + backupInfo.getStartTs() + ",completets="
384        + backupInfo.getCompleteTs() + ",bytescopied=" + backupInfo.getTotalBytesCopied();
385    if (LOG.isDebugEnabled()) {
386      LOG.debug("Backup " + backupInfo.getBackupId() + " finished: " + backupCompleteData);
387    }
388
389    // when full backup is done:
390    // - delete HBase snapshot
391    // - clean up directories with prefix "exportSnapshot-", which are generated when exporting
392    // snapshots
393    // incremental backups use distcp, which handles cleaning up its own directories
394    if (type == BackupType.FULL) {
395      deleteSnapshots(conn, backupInfo, conf);
396      cleanupExportSnapshotLog(conf);
397    }
398    BackupSystemTable.deleteSnapshot(conn);
399    backupManager.updateBackupInfo(backupInfo);
400
401    // Finish active session
402    backupManager.finishBackupSession();
403
404    LOG.info("Backup " + backupInfo.getBackupId() + " completed.");
405  }
406
407  /**
408   * Backup request execution.
409   * @throws IOException if the execution of the backup fails
410   */
411  public abstract void execute() throws IOException;
412
413  protected Stage getTestStage() {
414    return Stage.valueOf("stage_" + conf.getInt(BACKUP_TEST_MODE_STAGE, 0));
415  }
416
417  protected void failStageIf(Stage stage) throws IOException {
418    Stage current = getTestStage();
419    if (current == stage) {
420      throw new IOException("Failed stage " + stage + " in testing");
421    }
422  }
423
424  public enum Stage {
425    stage_0,
426    stage_1,
427    stage_2,
428    stage_3,
429    stage_4
430  }
431}