You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by el...@apache.org on 2017/03/20 22:49:39 UTC
[06/54] [abbrv] hbase git commit: HBASE-14123 HBase Backup/Restore
Phase 2 (Vladimir Rodionov)
http://git-wip-us.apache.org/repos/asf/hbase/blob/75d0f49d/hbase-server/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupManifest.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupManifest.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupManifest.java
new file mode 100644
index 0000000..b8adac9
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupManifest.java
@@ -0,0 +1,666 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.backup.impl;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.TreeMap;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.backup.BackupInfo;
+import org.apache.hadoop.hbase.backup.BackupType;
+import org.apache.hadoop.hbase.backup.util.BackupUtils;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.BackupProtos;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
+
+/**
+ * Backup manifest contains all the meta data of a backup image. The manifest info will be bundled
+ * as manifest file together with data. So that each backup image will contain all the info needed
+ * for restore. BackupManifest is a storage container for BackupImage.
+ * It is responsible for storing/reading backup image data and has some additional utility methods.
+ *
+ */
+@InterfaceAudience.Private
+public class BackupManifest {
+
+ private static final Log LOG = LogFactory.getLog(BackupManifest.class);
+
+ // manifest file name
+ public static final String MANIFEST_FILE_NAME = ".backup.manifest";
+
+ /**
+ * Backup image, the dependency graph is made up by series of backup images
+ * BackupImage contains all the relevant information to restore the backup and
+ * is used during restore operation
+ */
+
+ public static class BackupImage implements Comparable<BackupImage> {
+
+ static class Builder {
+ BackupImage image;
+
+ Builder() {
+ image = new BackupImage();
+ }
+
+ Builder withBackupId(String backupId) {
+ image.setBackupId(backupId);
+ return this;
+ }
+
+ Builder withType(BackupType type) {
+ image.setType(type);
+ return this;
+ }
+
+ Builder withRootDir(String rootDir) {
+ image.setRootDir(rootDir);
+ return this;
+ }
+
+ Builder withTableList(List<TableName> tableList) {
+ image.setTableList(tableList);
+ return this;
+ }
+
+ Builder withStartTime(long startTime) {
+ image.setStartTs(startTime);
+ return this;
+ }
+
+ Builder withCompleteTime(long completeTime) {
+ image.setCompleteTs(completeTime);
+ return this;
+ }
+
+ BackupImage build() {
+ return image;
+ }
+
+ }
+
+ private String backupId;
+ private BackupType type;
+ private String rootDir;
+ private List<TableName> tableList;
+ private long startTs;
+ private long completeTs;
+ private ArrayList<BackupImage> ancestors;
+ private HashMap<TableName, HashMap<String, Long>> incrTimeRanges;
+
+ static Builder newBuilder() {
+ return new Builder();
+ }
+
+ public BackupImage() {
+ super();
+ }
+
+ private BackupImage(String backupId, BackupType type, String rootDir,
+ List<TableName> tableList, long startTs, long completeTs) {
+ this.backupId = backupId;
+ this.type = type;
+ this.rootDir = rootDir;
+ this.tableList = tableList;
+ this.startTs = startTs;
+ this.completeTs = completeTs;
+ }
+
+ static BackupImage fromProto(BackupProtos.BackupImage im) {
+ String backupId = im.getBackupId();
+ String rootDir = im.getBackupRootDir();
+ long startTs = im.getStartTs();
+ long completeTs = im.getCompleteTs();
+ List<HBaseProtos.TableName> tableListList = im.getTableListList();
+ List<TableName> tableList = new ArrayList<TableName>();
+ for (HBaseProtos.TableName tn : tableListList) {
+ tableList.add(ProtobufUtil.toTableName(tn));
+ }
+
+ List<BackupProtos.BackupImage> ancestorList = im.getAncestorsList();
+
+ BackupType type =
+ im.getBackupType() == BackupProtos.BackupType.FULL ? BackupType.FULL
+ : BackupType.INCREMENTAL;
+
+ BackupImage image = new BackupImage(backupId, type, rootDir, tableList, startTs, completeTs);
+ for (BackupProtos.BackupImage img : ancestorList) {
+ image.addAncestor(fromProto(img));
+ }
+ image.setIncrTimeRanges(loadIncrementalTimestampMap(im));
+ return image;
+ }
+
+ BackupProtos.BackupImage toProto() {
+ BackupProtos.BackupImage.Builder builder = BackupProtos.BackupImage.newBuilder();
+ builder.setBackupId(backupId);
+ builder.setCompleteTs(completeTs);
+ builder.setStartTs(startTs);
+ builder.setBackupRootDir(rootDir);
+ if (type == BackupType.FULL) {
+ builder.setBackupType(BackupProtos.BackupType.FULL);
+ } else {
+ builder.setBackupType(BackupProtos.BackupType.INCREMENTAL);
+ }
+
+ for (TableName name : tableList) {
+ builder.addTableList(ProtobufUtil.toProtoTableName(name));
+ }
+
+ if (ancestors != null) {
+ for (BackupImage im : ancestors) {
+ builder.addAncestors(im.toProto());
+ }
+ }
+
+ setIncrementalTimestampMap(builder);
+ return builder.build();
+ }
+
+ private static HashMap<TableName, HashMap<String, Long>> loadIncrementalTimestampMap(
+ BackupProtos.BackupImage proto) {
+ List<BackupProtos.TableServerTimestamp> list = proto.getTstMapList();
+
+ HashMap<TableName, HashMap<String, Long>> incrTimeRanges =
+ new HashMap<TableName, HashMap<String, Long>>();
+ if (list == null || list.size() == 0) return incrTimeRanges;
+ for (BackupProtos.TableServerTimestamp tst : list) {
+ TableName tn = ProtobufUtil.toTableName(tst.getTableName());
+ HashMap<String, Long> map = incrTimeRanges.get(tn);
+ if (map == null) {
+ map = new HashMap<String, Long>();
+ incrTimeRanges.put(tn, map);
+ }
+ List<BackupProtos.ServerTimestamp> listSt = tst.getServerTimestampList();
+ for (BackupProtos.ServerTimestamp stm : listSt) {
+ ServerName sn = ProtobufUtil.toServerName(stm.getServerName());
+ map.put(sn.getHostname() + ":" + sn.getPort(), stm.getTimestamp());
+ }
+ }
+ return incrTimeRanges;
+ }
+
+ private void setIncrementalTimestampMap(BackupProtos.BackupImage.Builder builder) {
+ if (this.incrTimeRanges == null) {
+ return;
+ }
+ for (Entry<TableName, HashMap<String, Long>> entry : this.incrTimeRanges.entrySet()) {
+ TableName key = entry.getKey();
+ HashMap<String, Long> value = entry.getValue();
+ BackupProtos.TableServerTimestamp.Builder tstBuilder =
+ BackupProtos.TableServerTimestamp.newBuilder();
+ tstBuilder.setTableName(ProtobufUtil.toProtoTableName(key));
+
+ for (Map.Entry<String, Long> entry2 : value.entrySet()) {
+ String s = entry2.getKey();
+ BackupProtos.ServerTimestamp.Builder stBuilder =
+ BackupProtos.ServerTimestamp.newBuilder();
+ HBaseProtos.ServerName.Builder snBuilder = HBaseProtos.ServerName.newBuilder();
+ ServerName sn = ServerName.parseServerName(s);
+ snBuilder.setHostName(sn.getHostname());
+ snBuilder.setPort(sn.getPort());
+ stBuilder.setServerName(snBuilder.build());
+ stBuilder.setTimestamp(entry2.getValue());
+ tstBuilder.addServerTimestamp(stBuilder.build());
+ }
+ builder.addTstMap(tstBuilder.build());
+ }
+ }
+
+ public String getBackupId() {
+ return backupId;
+ }
+
+ private void setBackupId(String backupId) {
+ this.backupId = backupId;
+ }
+
+ public BackupType getType() {
+ return type;
+ }
+
+ private void setType(BackupType type) {
+ this.type = type;
+ }
+
+ public String getRootDir() {
+ return rootDir;
+ }
+
+ private void setRootDir(String rootDir) {
+ this.rootDir = rootDir;
+ }
+
+ public List<TableName> getTableNames() {
+ return tableList;
+ }
+
+ private void setTableList(List<TableName> tableList) {
+ this.tableList = tableList;
+ }
+
+ public long getStartTs() {
+ return startTs;
+ }
+
+ private void setStartTs(long startTs) {
+ this.startTs = startTs;
+ }
+
+ public long getCompleteTs() {
+ return completeTs;
+ }
+
+ private void setCompleteTs(long completeTs) {
+ this.completeTs = completeTs;
+ }
+
+ public ArrayList<BackupImage> getAncestors() {
+ if (this.ancestors == null) {
+ this.ancestors = new ArrayList<BackupImage>();
+ }
+ return this.ancestors;
+ }
+
+ private void addAncestor(BackupImage backupImage) {
+ this.getAncestors().add(backupImage);
+ }
+
+ public boolean hasAncestor(String token) {
+ for (BackupImage image : this.getAncestors()) {
+ if (image.getBackupId().equals(token)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ public boolean hasTable(TableName table) {
+ return tableList.contains(table);
+ }
+
+ @Override
+ public int compareTo(BackupImage other) {
+ String thisBackupId = this.getBackupId();
+ String otherBackupId = other.getBackupId();
+ int index1 = thisBackupId.lastIndexOf("_");
+ int index2 = otherBackupId.lastIndexOf("_");
+ String name1 = thisBackupId.substring(0, index1);
+ String name2 = otherBackupId.substring(0, index2);
+ if (name1.equals(name2)) {
+ Long thisTS = Long.valueOf(thisBackupId.substring(index1 + 1));
+ Long otherTS = Long.valueOf(otherBackupId.substring(index2 + 1));
+ return thisTS.compareTo(otherTS);
+ } else {
+ return name1.compareTo(name2);
+ }
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof BackupImage) {
+ return this.compareTo((BackupImage) obj) == 0;
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ int hash = 33 * this.getBackupId().hashCode() + type.hashCode();
+ hash = 33 * hash + rootDir.hashCode();
+ hash = 33 * hash + Long.valueOf(startTs).hashCode();
+ hash = 33 * hash + Long.valueOf(completeTs).hashCode();
+ for (TableName table : tableList) {
+ hash = 33 * hash + table.hashCode();
+ }
+ return hash;
+ }
+
+ public HashMap<TableName, HashMap<String, Long>> getIncrTimeRanges() {
+ return incrTimeRanges;
+ }
+
+ private void setIncrTimeRanges(HashMap<TableName, HashMap<String, Long>> incrTimeRanges) {
+ this.incrTimeRanges = incrTimeRanges;
+ }
+ }
+
+ // backup image directory
+ private String tableBackupDir = null;
+ private BackupImage backupImage;
+
+ /**
+ * Construct manifest for a ongoing backup.
+ * @param backup The ongoing backup info
+ */
+ public BackupManifest(BackupInfo backup) {
+
+ BackupImage.Builder builder = BackupImage.newBuilder();
+ this.backupImage =
+ builder.withBackupId(backup.getBackupId()).withType(backup.getType())
+ .withRootDir(backup.getBackupRootDir()).withTableList(backup.getTableNames())
+ .withStartTime(backup.getStartTs()).withCompleteTime(backup.getCompleteTs()).build();
+ }
+
+ /**
+ * Construct a table level manifest for a backup of the named table.
+ * @param backup The ongoing backup session info
+ */
+ public BackupManifest(BackupInfo backup, TableName table) {
+ this.tableBackupDir = backup.getTableBackupDir(table);
+ List<TableName> tables = new ArrayList<TableName>();
+ tables.add(table);
+ BackupImage.Builder builder = BackupImage.newBuilder();
+ this.backupImage =
+ builder.withBackupId(backup.getBackupId()).withType(backup.getType())
+ .withRootDir(backup.getBackupRootDir()).withTableList(tables)
+ .withStartTime(backup.getStartTs()).withCompleteTime(backup.getCompleteTs()).build();
+ }
+
+ /**
+ * Construct manifest from a backup directory.
+ * @param conf configuration
+ * @param backupPath backup path
+ * @throws IOException
+ */
+
+ public BackupManifest(Configuration conf, Path backupPath) throws IOException {
+ this(backupPath.getFileSystem(conf), backupPath);
+ }
+
+ /**
+ * Construct manifest from a backup directory.
+ * @param fs the FileSystem
+ * @param backupPath backup path
+ * @throws BackupException exception
+ */
+
+ public BackupManifest(FileSystem fs, Path backupPath) throws BackupException {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Loading manifest from: " + backupPath.toString());
+ }
+ // The input backupDir may not exactly be the backup table dir.
+ // It could be the backup log dir where there is also a manifest file stored.
+ // This variable's purpose is to keep the correct and original location so
+ // that we can store/persist it.
+ try {
+
+ FileStatus[] subFiles = BackupUtils.listStatus(fs, backupPath, null);
+ if (subFiles == null) {
+ String errorMsg = backupPath.toString() + " does not exist";
+ LOG.error(errorMsg);
+ throw new IOException(errorMsg);
+ }
+ for (FileStatus subFile : subFiles) {
+ if (subFile.getPath().getName().equals(MANIFEST_FILE_NAME)) {
+
+ // load and set manifest field from file content
+ FSDataInputStream in = fs.open(subFile.getPath());
+ long len = subFile.getLen();
+ byte[] pbBytes = new byte[(int) len];
+ in.readFully(pbBytes);
+ BackupProtos.BackupImage proto = null;
+ try {
+ proto = BackupProtos.BackupImage.parseFrom(pbBytes);
+ } catch (Exception e) {
+ throw new BackupException(e);
+ }
+ this.backupImage = BackupImage.fromProto(proto);
+ LOG.debug("Loaded manifest instance from manifest file: "
+ + BackupUtils.getPath(subFile.getPath()));
+ return;
+ }
+ }
+ String errorMsg = "No manifest file found in: " + backupPath.toString();
+ throw new IOException(errorMsg);
+
+ } catch (IOException e) {
+ throw new BackupException(e.getMessage());
+ }
+ }
+
+ public BackupType getType() {
+ return backupImage.getType();
+ }
+
+ /**
+ * Get the table set of this image.
+ * @return The table set list
+ */
+ public List<TableName> getTableList() {
+ return backupImage.getTableNames();
+ }
+
+ /**
+ * Persist the manifest file.
+ * @throws IOException IOException when storing the manifest file.
+ */
+
+ public void store(Configuration conf) throws BackupException {
+ byte[] data = backupImage.toProto().toByteArray();
+ // write the file, overwrite if already exist
+ String logBackupDir =
+ BackupUtils.getLogBackupDir(backupImage.getRootDir(), backupImage.getBackupId());
+ Path manifestFilePath =
+ new Path(new Path((tableBackupDir != null ? tableBackupDir : logBackupDir)),
+ MANIFEST_FILE_NAME);
+ try (FSDataOutputStream out =
+ manifestFilePath.getFileSystem(conf).create(manifestFilePath, true);) {
+ out.write(data);
+ } catch (IOException e) {
+ throw new BackupException(e.getMessage());
+ }
+
+ LOG.info("Manifest file stored to " + manifestFilePath);
+ }
+
+ /**
+ * Get this backup image.
+ * @return the backup image.
+ */
+ public BackupImage getBackupImage() {
+ return backupImage;
+ }
+
+ /**
+ * Add dependent backup image for this backup.
+ * @param image The direct dependent backup image
+ */
+ public void addDependentImage(BackupImage image) {
+ this.backupImage.addAncestor(image);
+ }
+
+ /**
+ * Set the incremental timestamp map directly.
+ * @param incrTimestampMap timestamp map
+ */
+ public void setIncrTimestampMap(HashMap<TableName, HashMap<String, Long>> incrTimestampMap) {
+ this.backupImage.setIncrTimeRanges(incrTimestampMap);
+ }
+
+ public Map<TableName, HashMap<String, Long>> getIncrTimestampMap() {
+ return backupImage.getIncrTimeRanges();
+ }
+
+ /**
+ * Get the image list of this backup for restore in time order.
+ * @param reverse If true, then output in reverse order, otherwise in time order from old to new
+ * @return the backup image list for restore in time order
+ */
+ public ArrayList<BackupImage> getRestoreDependentList(boolean reverse) {
+ TreeMap<Long, BackupImage> restoreImages = new TreeMap<Long, BackupImage>();
+ restoreImages.put(backupImage.startTs, backupImage);
+ for (BackupImage image : backupImage.getAncestors()) {
+ restoreImages.put(Long.valueOf(image.startTs), image);
+ }
+ return new ArrayList<BackupImage>(reverse ? (restoreImages.descendingMap().values())
+ : (restoreImages.values()));
+ }
+
+ /**
+ * Get the dependent image list for a specific table of this backup in time order from old to new
+ * if want to restore to this backup image level.
+ * @param table table
+ * @return the backup image list for a table in time order
+ */
+ public ArrayList<BackupImage> getDependentListByTable(TableName table) {
+ ArrayList<BackupImage> tableImageList = new ArrayList<BackupImage>();
+ ArrayList<BackupImage> imageList = getRestoreDependentList(true);
+ for (BackupImage image : imageList) {
+ if (image.hasTable(table)) {
+ tableImageList.add(image);
+ if (image.getType() == BackupType.FULL) {
+ break;
+ }
+ }
+ }
+ Collections.reverse(tableImageList);
+ return tableImageList;
+ }
+
+ /**
+ * Get the full dependent image list in the whole dependency scope for a specific table of this
+ * backup in time order from old to new.
+ * @param table table
+ * @return the full backup image list for a table in time order in the whole scope of the
+ * dependency of this image
+ */
+ public ArrayList<BackupImage> getAllDependentListByTable(TableName table) {
+ ArrayList<BackupImage> tableImageList = new ArrayList<BackupImage>();
+ ArrayList<BackupImage> imageList = getRestoreDependentList(false);
+ for (BackupImage image : imageList) {
+ if (image.hasTable(table)) {
+ tableImageList.add(image);
+ }
+ }
+ return tableImageList;
+ }
+
+ /**
+ * Check whether backup image1 could cover backup image2 or not.
+ * @param image1 backup image 1
+ * @param image2 backup image 2
+ * @return true if image1 can cover image2, otherwise false
+ */
+ public static boolean canCoverImage(BackupImage image1, BackupImage image2) {
+ // image1 can cover image2 only when the following conditions are satisfied:
+ // - image1 must not be an incremental image;
+ // - image1 must be taken after image2 has been taken;
+ // - table set of image1 must cover the table set of image2.
+ if (image1.getType() == BackupType.INCREMENTAL) {
+ return false;
+ }
+ if (image1.getStartTs() < image2.getStartTs()) {
+ return false;
+ }
+ List<TableName> image1TableList = image1.getTableNames();
+ List<TableName> image2TableList = image2.getTableNames();
+ boolean found = false;
+ for (int i = 0; i < image2TableList.size(); i++) {
+ found = false;
+ for (int j = 0; j < image1TableList.size(); j++) {
+ if (image2TableList.get(i).equals(image1TableList.get(j))) {
+ found = true;
+ break;
+ }
+ }
+ if (!found) {
+ return false;
+ }
+ }
+
+ LOG.debug("Backup image " + image1.getBackupId() + " can cover " + image2.getBackupId());
+ return true;
+ }
+
+ /**
+ * Check whether backup image set could cover a backup image or not.
+ * @param fullImages The backup image set
+ * @param image The target backup image
+ * @return true if fullImages can cover image, otherwise false
+ */
+ public static boolean canCoverImage(ArrayList<BackupImage> fullImages, BackupImage image) {
+ // fullImages can cover image only when the following conditions are satisfied:
+ // - each image of fullImages must not be an incremental image;
+ // - each image of fullImages must be taken after image has been taken;
+ // - sum table set of fullImages must cover the table set of image.
+ for (BackupImage image1 : fullImages) {
+ if (image1.getType() == BackupType.INCREMENTAL) {
+ return false;
+ }
+ if (image1.getStartTs() < image.getStartTs()) {
+ return false;
+ }
+ }
+
+ ArrayList<String> image1TableList = new ArrayList<String>();
+ for (BackupImage image1 : fullImages) {
+ List<TableName> tableList = image1.getTableNames();
+ for (TableName table : tableList) {
+ image1TableList.add(table.getNameAsString());
+ }
+ }
+ ArrayList<String> image2TableList = new ArrayList<String>();
+ List<TableName> tableList = image.getTableNames();
+ for (TableName table : tableList) {
+ image2TableList.add(table.getNameAsString());
+ }
+
+ for (int i = 0; i < image2TableList.size(); i++) {
+ if (image1TableList.contains(image2TableList.get(i)) == false) {
+ return false;
+ }
+ }
+
+ LOG.debug("Full image set can cover image " + image.getBackupId());
+ return true;
+ }
+
+ public BackupInfo toBackupInfo() {
+ BackupInfo info = new BackupInfo();
+ info.setType(backupImage.getType());
+ List<TableName> list = backupImage.getTableNames();
+ TableName[] tables = new TableName[list.size()];
+ info.addTables(list.toArray(tables));
+ info.setBackupId(backupImage.getBackupId());
+ info.setStartTs(backupImage.getStartTs());
+ info.setBackupRootDir(backupImage.getRootDir());
+ if (backupImage.getType() == BackupType.INCREMENTAL) {
+ info.setHLogTargetDir(BackupUtils.getLogBackupDir(backupImage.getRootDir(),
+ backupImage.getBackupId()));
+ }
+ return info;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/75d0f49d/hbase-server/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupSystemTable.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupSystemTable.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupSystemTable.java
new file mode 100644
index 0000000..6362f8e
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupSystemTable.java
@@ -0,0 +1,1376 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.backup.impl;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.backup.BackupInfo;
+import org.apache.hadoop.hbase.backup.BackupInfo.BackupState;
+import org.apache.hadoop.hbase.backup.BackupRestoreConstants;
+import org.apache.hadoop.hbase.backup.util.BackupUtils;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.BackupProtos;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+
+/**
+ * This class provides API to access backup system table<br>
+ *
+ * Backup system table schema:<br>
+ * <p><ul>
+ * <li>1. Backup sessions rowkey= "session:"+backupId; value =serialized BackupInfo</li>
+ * <li>2. Backup start code rowkey = "startcode:"+backupRoot; value = startcode</li>
+ * <li>3. Incremental backup set rowkey="incrbackupset:"+backupRoot; value=[list of tables]</li>
+ * <li>4. Table-RS-timestamp map rowkey="trslm:"+backupRoot+table_name;
+ * value = map[RS-> last WAL timestamp]</li>
+ * <li>5. RS - WAL ts map rowkey="rslogts:"+backupRoot +server; value = last WAL timestamp</li>
+ * <li>6. WALs recorded rowkey="wals:"+WAL unique file name;
+ * value = backupId and full WAL file name</li>
+ * </ul></p>
+ */
+@InterfaceAudience.Private
+public final class BackupSystemTable implements Closeable {
+
+ static class WALItem {
+ String backupId;
+ String walFile;
+ String backupRoot;
+
+ WALItem(String backupId, String walFile, String backupRoot) {
+ this.backupId = backupId;
+ this.walFile = walFile;
+ this.backupRoot = backupRoot;
+ }
+
+ public String getBackupId() {
+ return backupId;
+ }
+
+ public String getWalFile() {
+ return walFile;
+ }
+
+ public String getBackupRoot() {
+ return backupRoot;
+ }
+
+ @Override
+ public String toString() {
+ return Path.SEPARATOR + backupRoot + Path.SEPARATOR + backupId + Path.SEPARATOR + walFile;
+ }
+
+ }
+
+ private static final Log LOG = LogFactory.getLog(BackupSystemTable.class);
+
+ private TableName tableName;
+ /**
+ * Stores backup sessions (contexts)
+ */
+ final static byte[] SESSIONS_FAMILY = "session".getBytes();
+ /**
+ * Stores other meta
+ */
+ final static byte[] META_FAMILY = "meta".getBytes();
+ /**
+ * Connection to HBase cluster, shared among all instances
+ */
+ private final Connection connection;
+
+
+ private final static String BACKUP_INFO_PREFIX = "session:";
+ private final static String START_CODE_ROW = "startcode:";
+ private final static String INCR_BACKUP_SET = "incrbackupset:";
+ private final static String TABLE_RS_LOG_MAP_PREFIX = "trslm:";
+ private final static String RS_LOG_TS_PREFIX = "rslogts:";
+ private final static String WALS_PREFIX = "wals:";
+ private final static String SET_KEY_PREFIX = "backupset:";
+
+ private final static byte[] EMPTY_VALUE = new byte[] {};
+
+ // Safe delimiter in a string
+ private final static String NULL = "\u0000";
+
+ public BackupSystemTable(Connection conn) throws IOException {
+ this.connection = conn;
+ tableName = BackupSystemTable.getTableName(conn.getConfiguration());
+ checkSystemTable();
+ }
+
+ private void checkSystemTable() throws IOException {
+ try (Admin admin = connection.getAdmin();) {
+
+ if (!admin.tableExists(tableName)) {
+ HTableDescriptor backupHTD =
+ BackupSystemTable.getSystemTableDescriptor(connection.getConfiguration());
+ admin.createTable(backupHTD);
+ }
+ waitForSystemTable(admin);
+ }
+ }
+
+ private void waitForSystemTable(Admin admin) throws IOException {
+ long TIMEOUT = 60000;
+ long startTime = EnvironmentEdgeManager.currentTime();
+ while (!admin.tableExists(tableName) || !admin.isTableAvailable(tableName)) {
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ }
+ if (EnvironmentEdgeManager.currentTime() - startTime > TIMEOUT) {
+ throw new IOException("Failed to create backup system table after "+ TIMEOUT+"ms");
+ }
+ }
+ LOG.debug("Backup table exists and available");
+
+ }
+
+
+
+ @Override
+ public void close() {
+ // do nothing
+ }
+
+ /**
+ * Updates status (state) of a backup session in backup system table table
+ * @param info backup info
+ * @throws IOException exception
+ */
+ public void updateBackupInfo(BackupInfo info) throws IOException {
+
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("update backup status in backup system table for: " + info.getBackupId()
+ + " set status=" + info.getState());
+ }
+ try (Table table = connection.getTable(tableName)) {
+ Put put = createPutForBackupInfo(info);
+ table.put(put);
+ }
+ }
+
+ /**
+ * Deletes backup status from backup system table table
+ * @param backupId backup id
+ * @throws IOException exception
+ */
+
+ public void deleteBackupInfo(String backupId) throws IOException {
+
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("delete backup status in backup system table for " + backupId);
+ }
+ try (Table table = connection.getTable(tableName)) {
+ Delete del = createDeleteForBackupInfo(backupId);
+ table.delete(del);
+ }
+ }
+
+ /**
+ * Reads backup status object (instance of backup info) from backup system table table
+ * @param backupId backup id
+ * @return Current status of backup session or null
+ */
+
+ public BackupInfo readBackupInfo(String backupId) throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("read backup status from backup system table for: " + backupId);
+ }
+
+ try (Table table = connection.getTable(tableName)) {
+ Get get = createGetForBackupInfo(backupId);
+ Result res = table.get(get);
+ if (res.isEmpty()) {
+ return null;
+ }
+ return resultToBackupInfo(res);
+ }
+ }
+
+ /**
+ * Read the last backup start code (timestamp) of last successful backup. Will return null if
+ * there is no start code stored on hbase or the value is of length 0. These two cases indicate
+ * there is no successful backup completed so far.
+ * @param backupRoot directory path to backup destination
+ * @return the timestamp of last successful backup
+ * @throws IOException exception
+ */
+ public String readBackupStartCode(String backupRoot) throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("read backup start code from backup system table");
+ }
+ try (Table table = connection.getTable(tableName)) {
+ Get get = createGetForStartCode(backupRoot);
+ Result res = table.get(get);
+ if (res.isEmpty()) {
+ return null;
+ }
+ Cell cell = res.listCells().get(0);
+ byte[] val = CellUtil.cloneValue(cell);
+ if (val.length == 0) {
+ return null;
+ }
+ return new String(val);
+ }
+ }
+
+ /**
+ * Write the start code (timestamp) to backup system table. If passed in null, then write 0 byte.
+ * @param startCode start code
+ * @param backupRoot root directory path to backup
+ * @throws IOException exception
+ */
+ public void writeBackupStartCode(Long startCode, String backupRoot) throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("write backup start code to backup system table " + startCode);
+ }
+ try (Table table = connection.getTable(tableName)) {
+ Put put = createPutForStartCode(startCode.toString(), backupRoot);
+ table.put(put);
+ }
+ }
+
+ /**
+ * Get the Region Servers log information after the last log roll from backup system table.
+ * @param backupRoot root directory path to backup
+ * @return RS log info
+ * @throws IOException exception
+ */
+ public HashMap<String, Long> readRegionServerLastLogRollResult(String backupRoot)
+ throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("read region server last roll log result to backup system table");
+ }
+
+ Scan scan = createScanForReadRegionServerLastLogRollResult(backupRoot);
+
+ try (Table table = connection.getTable(tableName);
+ ResultScanner scanner = table.getScanner(scan)) {
+ Result res = null;
+ HashMap<String, Long> rsTimestampMap = new HashMap<String, Long>();
+ while ((res = scanner.next()) != null) {
+ res.advance();
+ Cell cell = res.current();
+ byte[] row = CellUtil.cloneRow(cell);
+ String server =
+ getServerNameForReadRegionServerLastLogRollResult(row);
+ byte[] data = CellUtil.cloneValue(cell);
+ rsTimestampMap.put(server, Bytes.toLong(data));
+ }
+ return rsTimestampMap;
+ }
+ }
+
+ /**
+ * Writes Region Server last roll log result (timestamp) to backup system table table
+ * @param server Region Server name
+ * @param ts last log timestamp
+ * @param backupRoot root directory path to backup
+ * @throws IOException exception
+ */
+ public void writeRegionServerLastLogRollResult(String server, Long ts, String backupRoot)
+ throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("write region server last roll log result to backup system table");
+ }
+ try (Table table = connection.getTable(tableName)) {
+ Put put =
+ createPutForRegionServerLastLogRollResult(server, ts, backupRoot);
+ table.put(put);
+ }
+ }
+
+ /**
+ * Get all completed backup information (in desc order by time)
+ * @param onlyCompleted true, if only successfully completed sessions
+ * @return history info of BackupCompleteData
+ * @throws IOException exception
+ */
+ public ArrayList<BackupInfo> getBackupHistory(boolean onlyCompleted) throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("get backup history from backup system table");
+ }
+ ArrayList<BackupInfo> list;
+ BackupState state = onlyCompleted ? BackupState.COMPLETE : BackupState.ANY;
+ list = getBackupInfos(state);
+ return BackupUtils.sortHistoryListDesc(list);
+ }
+
+ /**
+ * Get all backups history
+ * @return list of backup info
+ * @throws IOException
+ */
+ public List<BackupInfo> getBackupHistory() throws IOException {
+ return getBackupHistory(false);
+ }
+
+ /**
+ * Get first n backup history records
+ * @param n number of records
+ * @return list of records
+ * @throws IOException
+ */
+ public List<BackupInfo> getHistory(int n) throws IOException {
+
+ List<BackupInfo> history = getBackupHistory();
+ if (history.size() <= n) return history;
+ List<BackupInfo> list = new ArrayList<BackupInfo>();
+ for (int i = 0; i < n; i++) {
+ list.add(history.get(i));
+ }
+ return list;
+
+ }
+
+ /**
+ * Get backup history records filtered by list of filters.
+ * @param n max number of records
+ * @param filters list of filters
+ * @return backup records
+ * @throws IOException
+ */
+ public List<BackupInfo> getBackupHistory(int n, BackupInfo.Filter... filters) throws IOException {
+ if (filters.length == 0) return getHistory(n);
+
+ List<BackupInfo> history = getBackupHistory();
+ List<BackupInfo> result = new ArrayList<BackupInfo>();
+ for (BackupInfo bi : history) {
+ if (result.size() == n) break;
+ boolean passed = true;
+ for (int i = 0; i < filters.length; i++) {
+ if (!filters[i].apply(bi)) {
+ passed = false;
+ break;
+ }
+ }
+ if (passed) {
+ result.add(bi);
+ }
+ }
+ return result;
+
+ }
+
+ /**
+ * Get history for backup destination
+ * @param backupRoot backup destination path
+ * @return List of backup info
+ * @throws IOException
+ */
+ public List<BackupInfo> getBackupHistory(String backupRoot) throws IOException {
+ ArrayList<BackupInfo> history = getBackupHistory(false);
+ for (Iterator<BackupInfo> iterator = history.iterator(); iterator.hasNext();) {
+ BackupInfo info = iterator.next();
+ if (!backupRoot.equals(info.getBackupRootDir())) {
+ iterator.remove();
+ }
+ }
+ return history;
+ }
+
+ /**
+ * Get history for a table
+ * @param name table name
+ * @return history for a table
+ * @throws IOException
+ */
+ public List<BackupInfo> getBackupHistoryForTable(TableName name) throws IOException {
+ List<BackupInfo> history = getBackupHistory();
+ List<BackupInfo> tableHistory = new ArrayList<BackupInfo>();
+ for (BackupInfo info : history) {
+ List<TableName> tables = info.getTableNames();
+ if (tables.contains(name)) {
+ tableHistory.add(info);
+ }
+ }
+ return tableHistory;
+ }
+
+ public Map<TableName, ArrayList<BackupInfo>> getBackupHistoryForTableSet(Set<TableName> set,
+ String backupRoot) throws IOException {
+ List<BackupInfo> history = getBackupHistory(backupRoot);
+ Map<TableName, ArrayList<BackupInfo>> tableHistoryMap =
+ new HashMap<TableName, ArrayList<BackupInfo>>();
+ for (Iterator<BackupInfo> iterator = history.iterator(); iterator.hasNext();) {
+ BackupInfo info = iterator.next();
+ if (!backupRoot.equals(info.getBackupRootDir())) {
+ continue;
+ }
+ List<TableName> tables = info.getTableNames();
+ for (TableName tableName : tables) {
+ if (set.contains(tableName)) {
+ ArrayList<BackupInfo> list = tableHistoryMap.get(tableName);
+ if (list == null) {
+ list = new ArrayList<BackupInfo>();
+ tableHistoryMap.put(tableName, list);
+ }
+ list.add(info);
+ }
+ }
+ }
+ return tableHistoryMap;
+ }
+
+ /**
+ * Get all backup sessions with a given state (in descending order by time)
+ * @param state backup session state
+ * @return history info of backup info objects
+ * @throws IOException exception
+ */
+ public ArrayList<BackupInfo> getBackupInfos(BackupState state) throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("get backup infos from backup system table");
+ }
+
+ Scan scan = createScanForBackupHistory();
+ ArrayList<BackupInfo> list = new ArrayList<BackupInfo>();
+
+ try (Table table = connection.getTable(tableName);
+ ResultScanner scanner = table.getScanner(scan)) {
+ Result res = null;
+ while ((res = scanner.next()) != null) {
+ res.advance();
+ BackupInfo context = cellToBackupInfo(res.current());
+ if (state != BackupState.ANY && context.getState() != state) {
+ continue;
+ }
+ list.add(context);
+ }
+ return list;
+ }
+ }
+
+ /**
+ * Write the current timestamps for each regionserver to backup system table after a successful
+ * full or incremental backup. The saved timestamp is of the last log file that was backed up
+ * already.
+ * @param tables tables
+ * @param newTimestamps timestamps
+ * @param backupRoot root directory path to backup
+ * @throws IOException exception
+ */
+ public void writeRegionServerLogTimestamp(Set<TableName> tables,
+ HashMap<String, Long> newTimestamps, String backupRoot) throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("write RS log time stamps to backup system table for tables ["
+ + StringUtils.join(tables, ",") + "]");
+ }
+ List<Put> puts = new ArrayList<Put>();
+ for (TableName table : tables) {
+ byte[] smapData = toTableServerTimestampProto(table, newTimestamps).toByteArray();
+ Put put =
+ createPutForWriteRegionServerLogTimestamp(table, smapData,
+ backupRoot);
+ puts.add(put);
+ }
+ try (Table table = connection.getTable(tableName)) {
+ table.put(puts);
+ }
+ }
+
+ /**
+ * Read the timestamp for each region server log after the last successful backup. Each table has
+ * its own set of the timestamps. The info is stored for each table as a concatenated string of
+ * rs->timestapmp
+ * @param backupRoot root directory path to backup
+ * @return the timestamp for each region server. key: tableName value:
+ * RegionServer,PreviousTimeStamp
+ * @throws IOException exception
+ */
+ public HashMap<TableName, HashMap<String, Long>> readLogTimestampMap(String backupRoot)
+ throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("read RS log ts from backup system table for root=" + backupRoot);
+ }
+
+ HashMap<TableName, HashMap<String, Long>> tableTimestampMap =
+ new HashMap<TableName, HashMap<String, Long>>();
+
+ Scan scan = createScanForReadLogTimestampMap(backupRoot);
+ try (Table table = connection.getTable(tableName);
+ ResultScanner scanner = table.getScanner(scan)) {
+ Result res = null;
+ while ((res = scanner.next()) != null) {
+ res.advance();
+ Cell cell = res.current();
+ byte[] row = CellUtil.cloneRow(cell);
+ String tabName = getTableNameForReadLogTimestampMap(row);
+ TableName tn = TableName.valueOf(tabName);
+ byte[] data = CellUtil.cloneValue(cell);
+ if (data == null) {
+ throw new IOException("Data of last backup data from backup system table "
+ + "is empty. Create a backup first.");
+ }
+ if (data != null && data.length > 0) {
+ HashMap<String, Long> lastBackup =
+ fromTableServerTimestampProto(BackupProtos.TableServerTimestamp.parseFrom(data));
+ tableTimestampMap.put(tn, lastBackup);
+ }
+ }
+ return tableTimestampMap;
+ }
+ }
+
+ private BackupProtos.TableServerTimestamp toTableServerTimestampProto(TableName table,
+ Map<String, Long> map) {
+ BackupProtos.TableServerTimestamp.Builder tstBuilder =
+ BackupProtos.TableServerTimestamp.newBuilder();
+ tstBuilder.setTableName(org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil
+ .toProtoTableName(table));
+
+ for (Entry<String, Long> entry : map.entrySet()) {
+ BackupProtos.ServerTimestamp.Builder builder = BackupProtos.ServerTimestamp.newBuilder();
+ HBaseProtos.ServerName.Builder snBuilder = HBaseProtos.ServerName.newBuilder();
+ ServerName sn = ServerName.parseServerName(entry.getKey());
+ snBuilder.setHostName(sn.getHostname());
+ snBuilder.setPort(sn.getPort());
+ builder.setServerName(snBuilder.build());
+ builder.setTimestamp(entry.getValue());
+ tstBuilder.addServerTimestamp(builder.build());
+ }
+
+ return tstBuilder.build();
+ }
+
+ private HashMap<String, Long> fromTableServerTimestampProto(
+ BackupProtos.TableServerTimestamp proto) {
+ HashMap<String, Long> map = new HashMap<String, Long>();
+ List<BackupProtos.ServerTimestamp> list = proto.getServerTimestampList();
+ for (BackupProtos.ServerTimestamp st : list) {
+ ServerName sn =
+ org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil.toServerName(st.getServerName());
+ map.put(sn.getHostname() + ":" + sn.getPort(), st.getTimestamp());
+ }
+ return map;
+ }
+
+ /**
+ * Return the current tables covered by incremental backup.
+ * @param backupRoot root directory path to backup
+ * @return set of tableNames
+ * @throws IOException exception
+ */
+ public Set<TableName> getIncrementalBackupTableSet(String backupRoot) throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("get incremental backup table set from backup system table");
+ }
+ TreeSet<TableName> set = new TreeSet<>();
+
+ try (Table table = connection.getTable(tableName)) {
+ Get get = createGetForIncrBackupTableSet(backupRoot);
+ Result res = table.get(get);
+ if (res.isEmpty()) {
+ return set;
+ }
+ List<Cell> cells = res.listCells();
+ for (Cell cell : cells) {
+ // qualifier = table name - we use table names as qualifiers
+ set.add(TableName.valueOf(CellUtil.cloneQualifier(cell)));
+ }
+ return set;
+ }
+ }
+
+ /**
+ * Add tables to global incremental backup set
+ * @param tables set of tables
+ * @param backupRoot root directory path to backup
+ * @throws IOException exception
+ */
+ public void addIncrementalBackupTableSet(Set<TableName> tables, String backupRoot)
+ throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Add incremental backup table set to backup system table. ROOT=" + backupRoot
+ + " tables [" + StringUtils.join(tables, " ") + "]");
+ for (TableName table : tables) {
+ LOG.debug(table);
+ }
+ }
+ try (Table table = connection.getTable(tableName)) {
+ Put put = createPutForIncrBackupTableSet(tables, backupRoot);
+ table.put(put);
+ }
+ }
+
+ /**
+ * Deletes incremental backup set for a backup destination
+ * @param backupRoot backup root
+ */
+
+ public void deleteIncrementalBackupTableSet(String backupRoot) throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Delete incremental backup table set to backup system table. ROOT=" + backupRoot);
+ }
+ try (Table table = connection.getTable(tableName)) {
+ Delete delete = createDeleteForIncrBackupTableSet(backupRoot);
+ table.delete(delete);
+ }
+ }
+
+ /**
+ * Register WAL files as eligible for deletion
+ * @param files files
+ * @param backupId backup id
+ * @param backupRoot root directory path to backup destination
+ * @throws IOException exception
+ */
+ public void addWALFiles(List<String> files, String backupId, String backupRoot)
+ throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("add WAL files to backup system table: " + backupId + " " + backupRoot + " files ["
+ + StringUtils.join(files, ",") + "]");
+ for (String f : files) {
+ LOG.debug("add :" + f);
+ }
+ }
+ try (Table table = connection.getTable(tableName)) {
+ List<Put> puts =
+ createPutsForAddWALFiles(files, backupId, backupRoot);
+ table.put(puts);
+ }
+ }
+
+ /**
+ * Register WAL files as eligible for deletion
+ * @param backupRoot root directory path to backup
+ * @throws IOException exception
+ */
+ public Iterator<WALItem> getWALFilesIterator(String backupRoot) throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("get WAL files from backup system table");
+ }
+ final Table table = connection.getTable(tableName);
+ Scan scan = createScanForGetWALs(backupRoot);
+ final ResultScanner scanner = table.getScanner(scan);
+ final Iterator<Result> it = scanner.iterator();
+ return new Iterator<WALItem>() {
+
+ @Override
+ public boolean hasNext() {
+ boolean next = it.hasNext();
+ if (!next) {
+ // close all
+ try {
+ scanner.close();
+ table.close();
+ } catch (IOException e) {
+ LOG.error("Close WAL Iterator", e);
+ }
+ }
+ return next;
+ }
+
+ @Override
+ public WALItem next() {
+ Result next = it.next();
+ List<Cell> cells = next.listCells();
+ byte[] buf = cells.get(0).getValueArray();
+ int len = cells.get(0).getValueLength();
+ int offset = cells.get(0).getValueOffset();
+ String backupId = new String(buf, offset, len);
+ buf = cells.get(1).getValueArray();
+ len = cells.get(1).getValueLength();
+ offset = cells.get(1).getValueOffset();
+ String walFile = new String(buf, offset, len);
+ buf = cells.get(2).getValueArray();
+ len = cells.get(2).getValueLength();
+ offset = cells.get(2).getValueOffset();
+ String backupRoot = new String(buf, offset, len);
+ return new WALItem(backupId, walFile, backupRoot);
+ }
+
+ @Override
+ public void remove() {
+ // not implemented
+ throw new RuntimeException("remove is not supported");
+ }
+ };
+
+ }
+
+ /**
+ * Check if WAL file is eligible for deletion Future: to support all backup destinations
+ * @param file name of a file to check
+ * @return true, if deletable, false otherwise.
+ * @throws IOException exception
+ */
+ public boolean isWALFileDeletable(String file) throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Check if WAL file has been already backed up in backup system table " + file);
+ }
+ try (Table table = connection.getTable(tableName)) {
+ Get get = createGetForCheckWALFile(file);
+ Result res = table.get(get);
+ if (res.isEmpty()) {
+ return false;
+ }
+ return true;
+ }
+ }
+
+ /**
+ * Checks if we have at least one backup session in backup system table This API is used by
+ * BackupLogCleaner
+ * @return true, if - at least one session exists in backup system table table
+ * @throws IOException exception
+ */
+ public boolean hasBackupSessions() throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Has backup sessions from backup system table");
+ }
+ boolean result = false;
+ Scan scan = createScanForBackupHistory();
+ scan.setCaching(1);
+ try (Table table = connection.getTable(tableName);
+ ResultScanner scanner = table.getScanner(scan)) {
+ if (scanner.next() != null) {
+ result = true;
+ }
+ return result;
+ }
+ }
+
+ /**
+ * BACKUP SETS
+ */
+
+ /**
+ * Get backup set list
+ * @return backup set list
+ * @throws IOException
+ */
+ public List<String> listBackupSets() throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(" Backup set list");
+ }
+ List<String> list = new ArrayList<String>();
+ Table table = null;
+ ResultScanner scanner = null;
+ try {
+ table = connection.getTable(tableName);
+ Scan scan = createScanForBackupSetList();
+ scan.setMaxVersions(1);
+ scanner = table.getScanner(scan);
+ Result res = null;
+ while ((res = scanner.next()) != null) {
+ res.advance();
+ list.add(cellKeyToBackupSetName(res.current()));
+ }
+ return list;
+ } finally {
+ if (scanner != null) {
+ scanner.close();
+ }
+ if (table != null) {
+ table.close();
+ }
+ }
+ }
+
+ /**
+ * Get backup set description (list of tables)
+ * @param name set's name
+ * @return list of tables in a backup set
+ * @throws IOException
+ */
+ public List<TableName> describeBackupSet(String name) throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(" Backup set describe: " + name);
+ }
+ Table table = null;
+ try {
+ table = connection.getTable(tableName);
+ Get get = createGetForBackupSet(name);
+ Result res = table.get(get);
+ if (res.isEmpty()) return null;
+ res.advance();
+ String[] tables = cellValueToBackupSet(res.current());
+ return toList(tables);
+ } finally {
+ if (table != null) {
+ table.close();
+ }
+ }
+ }
+
+ private List<TableName> toList(String[] tables) {
+ List<TableName> list = new ArrayList<TableName>(tables.length);
+ for (String name : tables) {
+ list.add(TableName.valueOf(name));
+ }
+ return list;
+ }
+
+ /**
+ * Add backup set (list of tables)
+ * @param name set name
+ * @param newTables list of tables, comma-separated
+ * @throws IOException
+ */
+ public void addToBackupSet(String name, String[] newTables) throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Backup set add: " + name + " tables [" + StringUtils.join(newTables, " ") + "]");
+ }
+ Table table = null;
+ String[] union = null;
+ try {
+ table = connection.getTable(tableName);
+ Get get = createGetForBackupSet(name);
+ Result res = table.get(get);
+ if (res.isEmpty()) {
+ union = newTables;
+ } else {
+ res.advance();
+ String[] tables = cellValueToBackupSet(res.current());
+ union = merge(tables, newTables);
+ }
+ Put put = createPutForBackupSet(name, union);
+ table.put(put);
+ } finally {
+ if (table != null) {
+ table.close();
+ }
+ }
+ }
+
+ private String[] merge(String[] tables, String[] newTables) {
+ List<String> list = new ArrayList<String>();
+ // Add all from tables
+ for (String t : tables) {
+ list.add(t);
+ }
+ for (String nt : newTables) {
+ if (list.contains(nt)) continue;
+ list.add(nt);
+ }
+ String[] arr = new String[list.size()];
+ list.toArray(arr);
+ return arr;
+ }
+
+ /**
+ * Remove tables from backup set (list of tables)
+ * @param name set name
+ * @param toRemove list of tables
+ * @throws IOException
+ */
+ public void removeFromBackupSet(String name, String[] toRemove) throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(" Backup set remove from : " + name + " tables [" + StringUtils.join(toRemove, " ")
+ + "]");
+ }
+ Table table = null;
+ String[] disjoint = null;
+ String[] tables = null;
+ try {
+ table = connection.getTable(tableName);
+ Get get = createGetForBackupSet(name);
+ Result res = table.get(get);
+ if (res.isEmpty()) {
+ LOG.warn("Backup set '" + name + "' not found.");
+ return;
+ } else {
+ res.advance();
+ tables = cellValueToBackupSet(res.current());
+ disjoint = disjoin(tables, toRemove);
+ }
+ if (disjoint.length > 0 && disjoint.length != tables.length) {
+ Put put = createPutForBackupSet(name, disjoint);
+ table.put(put);
+ } else if(disjoint.length == tables.length) {
+ LOG.warn("Backup set '" + name + "' does not contain tables ["
+ + StringUtils.join(toRemove, " ") + "]");
+ } else { // disjoint.length == 0 and tables.length >0
+ // Delete backup set
+ LOG.info("Backup set '"+name+"' is empty. Deleting.");
+ deleteBackupSet(name);
+ }
+ } finally {
+ if (table != null) {
+ table.close();
+ }
+ }
+ }
+
+ private String[] disjoin(String[] tables, String[] toRemove) {
+ List<String> list = new ArrayList<String>();
+ // Add all from tables
+ for (String t : tables) {
+ list.add(t);
+ }
+ for (String nt : toRemove) {
+ if (list.contains(nt)) {
+ list.remove(nt);
+ }
+ }
+ String[] arr = new String[list.size()];
+ list.toArray(arr);
+ return arr;
+ }
+
+ /**
+ * Delete backup set
+ * @param name set's name
+ * @throws IOException
+ */
+ public void deleteBackupSet(String name) throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(" Backup set delete: " + name);
+ }
+ Table table = null;
+ try {
+ table = connection.getTable(tableName);
+ Delete del = createDeleteForBackupSet(name);
+ table.delete(del);
+ } finally {
+ if (table != null) {
+ table.close();
+ }
+ }
+ }
+
+ /**
+ * Get backup system table descriptor
+ * @return table's descriptor
+ */
+ public static HTableDescriptor getSystemTableDescriptor(Configuration conf) {
+
+ HTableDescriptor tableDesc = new HTableDescriptor(getTableName(conf));
+ HColumnDescriptor colSessionsDesc = new HColumnDescriptor(SESSIONS_FAMILY);
+ colSessionsDesc.setMaxVersions(1);
+ // Time to keep backup sessions (secs)
+ Configuration config = HBaseConfiguration.create();
+ int ttl =
+ config.getInt(BackupRestoreConstants.BACKUP_SYSTEM_TTL_KEY,
+ BackupRestoreConstants.BACKUP_SYSTEM_TTL_DEFAULT);
+ colSessionsDesc.setTimeToLive(ttl);
+ tableDesc.addFamily(colSessionsDesc);
+ HColumnDescriptor colMetaDesc = new HColumnDescriptor(META_FAMILY);
+ tableDesc.addFamily(colMetaDesc);
+ return tableDesc;
+ }
+
+ public static TableName getTableName(Configuration conf) {
+ String name =
+ conf.get(BackupRestoreConstants.BACKUP_SYSTEM_TABLE_NAME_KEY,
+ BackupRestoreConstants.BACKUP_SYSTEM_TABLE_NAME_DEFAULT);
+ return TableName.valueOf(name);
+ }
+
+ public static String getTableNameAsString(Configuration conf) {
+ return getTableName(conf).getNameAsString();
+ }
+
+
+
+
+
+ /**
+ * Creates Put operation for a given backup info object
+ * @param context backup info
+ * @return put operation
+ * @throws IOException exception
+ */
+ private Put createPutForBackupInfo(BackupInfo context) throws IOException {
+ Put put = new Put(rowkey(BACKUP_INFO_PREFIX, context.getBackupId()));
+ put.addColumn(BackupSystemTable.SESSIONS_FAMILY, Bytes.toBytes("context"),
+ context.toByteArray());
+ return put;
+ }
+
+ /**
+ * Creates Get operation for a given backup id
+ * @param backupId backup's ID
+ * @return get operation
+ * @throws IOException exception
+ */
+ private Get createGetForBackupInfo(String backupId) throws IOException {
+ Get get = new Get(rowkey(BACKUP_INFO_PREFIX, backupId));
+ get.addFamily(BackupSystemTable.SESSIONS_FAMILY);
+ get.setMaxVersions(1);
+ return get;
+ }
+
+ /**
+ * Creates Delete operation for a given backup id
+ * @param backupId backup's ID
+ * @return delete operation
+ * @throws IOException exception
+ */
+ private Delete createDeleteForBackupInfo(String backupId) {
+ Delete del = new Delete(rowkey(BACKUP_INFO_PREFIX, backupId));
+ del.addFamily(BackupSystemTable.SESSIONS_FAMILY);
+ return del;
+ }
+
+ /**
+ * Converts Result to BackupInfo
+ * @param res HBase result
+ * @return backup info instance
+ * @throws IOException exception
+ */
+ private BackupInfo resultToBackupInfo(Result res) throws IOException {
+ res.advance();
+ Cell cell = res.current();
+ return cellToBackupInfo(cell);
+ }
+
+ /**
+ * Creates Get operation to retrieve start code from backup system table
+ * @return get operation
+ * @throws IOException exception
+ */
+ private Get createGetForStartCode(String rootPath) throws IOException {
+ Get get = new Get(rowkey(START_CODE_ROW, rootPath));
+ get.addFamily(BackupSystemTable.META_FAMILY);
+ get.setMaxVersions(1);
+ return get;
+ }
+
+ /**
+ * Creates Put operation to store start code to backup system table
+ * @return put operation
+ * @throws IOException exception
+ */
+ private Put createPutForStartCode(String startCode, String rootPath) {
+ Put put = new Put(rowkey(START_CODE_ROW, rootPath));
+ put.addColumn(BackupSystemTable.META_FAMILY, Bytes.toBytes("startcode"),
+ Bytes.toBytes(startCode));
+ return put;
+ }
+
+ /**
+ * Creates Get to retrieve incremental backup table set from backup system table
+ * @return get operation
+ * @throws IOException exception
+ */
+ private Get createGetForIncrBackupTableSet(String backupRoot) throws IOException {
+ Get get = new Get(rowkey(INCR_BACKUP_SET, backupRoot));
+ get.addFamily(BackupSystemTable.META_FAMILY);
+ get.setMaxVersions(1);
+ return get;
+ }
+
+ /**
+ * Creates Put to store incremental backup table set
+ * @param tables tables
+ * @return put operation
+ */
+ private Put createPutForIncrBackupTableSet(Set<TableName> tables, String backupRoot) {
+ Put put = new Put(rowkey(INCR_BACKUP_SET, backupRoot));
+ for (TableName table : tables) {
+ put.addColumn(BackupSystemTable.META_FAMILY, Bytes.toBytes(table.getNameAsString()),
+ EMPTY_VALUE);
+ }
+ return put;
+ }
+
+ /**
+ * Creates Delete for incremental backup table set
+ * @param backupRoot backup root
+ * @return delete operation
+ */
+ private Delete createDeleteForIncrBackupTableSet(String backupRoot) {
+ Delete delete = new Delete(rowkey(INCR_BACKUP_SET, backupRoot));
+ delete.addFamily(BackupSystemTable.META_FAMILY);
+ return delete;
+ }
+
+ /**
+ * Creates Scan operation to load backup history
+ * @return scan operation
+ */
+ private Scan createScanForBackupHistory() {
+ Scan scan = new Scan();
+ byte[] startRow = Bytes.toBytes(BACKUP_INFO_PREFIX);
+ byte[] stopRow = Arrays.copyOf(startRow, startRow.length);
+ stopRow[stopRow.length - 1] = (byte) (stopRow[stopRow.length - 1] + 1);
+ scan.setStartRow(startRow);
+ scan.setStopRow(stopRow);
+ scan.addFamily(BackupSystemTable.SESSIONS_FAMILY);
+ scan.setMaxVersions(1);
+ return scan;
+ }
+
+ /**
+ * Converts cell to backup info instance.
+ * @param current current cell
+ * @return backup backup info instance
+ * @throws IOException exception
+ */
+ private BackupInfo cellToBackupInfo(Cell current) throws IOException {
+ byte[] data = CellUtil.cloneValue(current);
+ return BackupInfo.fromByteArray(data);
+ }
+
+ /**
+ * Creates Put to write RS last roll log timestamp map
+ * @param table table
+ * @param smap map, containing RS:ts
+ * @return put operation
+ */
+ private Put createPutForWriteRegionServerLogTimestamp(TableName table, byte[] smap,
+ String backupRoot) {
+ Put put = new Put(rowkey(TABLE_RS_LOG_MAP_PREFIX, backupRoot, NULL, table.getNameAsString()));
+ put.addColumn(BackupSystemTable.META_FAMILY, Bytes.toBytes("log-roll-map"), smap);
+ return put;
+ }
+
+ /**
+ * Creates Scan to load table-> { RS -> ts} map of maps
+ * @return scan operation
+ */
+ private Scan createScanForReadLogTimestampMap(String backupRoot) {
+ Scan scan = new Scan();
+ byte[] startRow = rowkey(TABLE_RS_LOG_MAP_PREFIX, backupRoot);
+ byte[] stopRow = Arrays.copyOf(startRow, startRow.length);
+ stopRow[stopRow.length - 1] = (byte) (stopRow[stopRow.length - 1] + 1);
+ scan.setStartRow(startRow);
+ scan.setStopRow(stopRow);
+ scan.addFamily(BackupSystemTable.META_FAMILY);
+
+ return scan;
+ }
+
+ /**
+ * Get table name from rowkey
+ * @param cloneRow rowkey
+ * @return table name
+ */
+ private String getTableNameForReadLogTimestampMap(byte[] cloneRow) {
+ String s = Bytes.toString(cloneRow);
+ int index = s.lastIndexOf(NULL);
+ return s.substring(index + 1);
+ }
+
+ /**
+ * Creates Put to store RS last log result
+ * @param server server name
+ * @param timestamp log roll result (timestamp)
+ * @return put operation
+ */
+ private Put createPutForRegionServerLastLogRollResult(String server, Long timestamp,
+ String backupRoot) {
+ Put put = new Put(rowkey(RS_LOG_TS_PREFIX, backupRoot, NULL, server));
+ put.addColumn(BackupSystemTable.META_FAMILY, Bytes.toBytes("rs-log-ts"),
+ Bytes.toBytes(timestamp));
+ return put;
+ }
+
+ /**
+ * Creates Scan operation to load last RS log roll results
+ * @return scan operation
+ */
+ private Scan createScanForReadRegionServerLastLogRollResult(String backupRoot) {
+ Scan scan = new Scan();
+ byte[] startRow = rowkey(RS_LOG_TS_PREFIX, backupRoot);
+ byte[] stopRow = Arrays.copyOf(startRow, startRow.length);
+ stopRow[stopRow.length - 1] = (byte) (stopRow[stopRow.length - 1] + 1);
+ scan.setStartRow(startRow);
+ scan.setStopRow(stopRow);
+ scan.addFamily(BackupSystemTable.META_FAMILY);
+ scan.setMaxVersions(1);
+
+ return scan;
+ }
+
+ /**
+ * Get server's name from rowkey
+ * @param row rowkey
+ * @return server's name
+ */
+ private String getServerNameForReadRegionServerLastLogRollResult(byte[] row) {
+ String s = Bytes.toString(row);
+ int index = s.lastIndexOf(NULL);
+ return s.substring(index + 1);
+ }
+
+ /**
+ * Creates put list for list of WAL files
+ * @param files list of WAL file paths
+ * @param backupId backup id
+ * @return put list
+ * @throws IOException exception
+ */
+ private List<Put> createPutsForAddWALFiles(List<String> files, String backupId,
+ String backupRoot) throws IOException {
+
+ List<Put> puts = new ArrayList<Put>();
+ for (String file : files) {
+ Put put = new Put(rowkey(WALS_PREFIX, BackupUtils.getUniqueWALFileNamePart(file)));
+ put.addColumn(BackupSystemTable.META_FAMILY, Bytes.toBytes("backupId"),
+ Bytes.toBytes(backupId));
+ put.addColumn(BackupSystemTable.META_FAMILY, Bytes.toBytes("file"), Bytes.toBytes(file));
+ put.addColumn(BackupSystemTable.META_FAMILY, Bytes.toBytes("root"), Bytes.toBytes(backupRoot));
+ puts.add(put);
+ }
+ return puts;
+ }
+
+ /**
+ * Creates Scan operation to load WALs
+ * @param backupRoot path to backup destination
+ * @return scan operation
+ */
+ private Scan createScanForGetWALs(String backupRoot) {
+ // TODO: support for backupRoot
+ Scan scan = new Scan();
+ byte[] startRow = Bytes.toBytes(WALS_PREFIX);
+ byte[] stopRow = Arrays.copyOf(startRow, startRow.length);
+ stopRow[stopRow.length - 1] = (byte) (stopRow[stopRow.length - 1] + 1);
+ scan.setStartRow(startRow);
+ scan.setStopRow(stopRow);
+ scan.addFamily(BackupSystemTable.META_FAMILY);
+ return scan;
+ }
+
+ /**
+ * Creates Get operation for a given wal file name TODO: support for backup destination
+ * @param file file
+ * @return get operation
+ * @throws IOException exception
+ */
+ private Get createGetForCheckWALFile(String file) throws IOException {
+ Get get = new Get(rowkey(WALS_PREFIX, BackupUtils.getUniqueWALFileNamePart(file)));
+ // add backup root column
+ get.addFamily(BackupSystemTable.META_FAMILY);
+ return get;
+ }
+
+ /**
+ * Creates Scan operation to load backup set list
+ * @return scan operation
+ */
+ private Scan createScanForBackupSetList() {
+ Scan scan = new Scan();
+ byte[] startRow = Bytes.toBytes(SET_KEY_PREFIX);
+ byte[] stopRow = Arrays.copyOf(startRow, startRow.length);
+ stopRow[stopRow.length - 1] = (byte) (stopRow[stopRow.length - 1] + 1);
+ scan.setStartRow(startRow);
+ scan.setStopRow(stopRow);
+ scan.addFamily(BackupSystemTable.META_FAMILY);
+ return scan;
+ }
+
+ /**
+ * Creates Get operation to load backup set content
+ * @return get operation
+ */
+ private Get createGetForBackupSet(String name) {
+ Get get = new Get(rowkey(SET_KEY_PREFIX, name));
+ get.addFamily(BackupSystemTable.META_FAMILY);
+ return get;
+ }
+
+ /**
+ * Creates Delete operation to delete backup set content
+ * @param name backup set's name
+ * @return delete operation
+ */
+ private Delete createDeleteForBackupSet(String name) {
+ Delete del = new Delete(rowkey(SET_KEY_PREFIX, name));
+ del.addFamily(BackupSystemTable.META_FAMILY);
+ return del;
+ }
+
+ /**
+ * Creates Put operation to update backup set content
+ * @param name backup set's name
+ * @param tables list of tables
+ * @return put operation
+ */
+ private Put createPutForBackupSet(String name, String[] tables) {
+ Put put = new Put(rowkey(SET_KEY_PREFIX, name));
+ byte[] value = convertToByteArray(tables);
+ put.addColumn(BackupSystemTable.META_FAMILY, Bytes.toBytes("tables"), value);
+ return put;
+ }
+
+ private byte[] convertToByteArray(String[] tables) {
+ return StringUtils.join(tables, ",").getBytes();
+ }
+
+ /**
+ * Converts cell to backup set list.
+ * @param current current cell
+ * @return backup set as array of table names
+ * @throws IOException
+ */
+ private String[] cellValueToBackupSet(Cell current) throws IOException {
+ byte[] data = CellUtil.cloneValue(current);
+ if (data != null && data.length > 0) {
+ return Bytes.toString(data).split(",");
+ } else {
+ return new String[0];
+ }
+ }
+
+ /**
+ * Converts cell key to backup set name.
+ * @param current current cell
+ * @return backup set name
+ * @throws IOException
+ */
+ private String cellKeyToBackupSetName(Cell current) throws IOException {
+ byte[] data = CellUtil.cloneRow(current);
+ return Bytes.toString(data).substring(SET_KEY_PREFIX.length());
+ }
+
+ private byte[] rowkey(String s, String... other) {
+ StringBuilder sb = new StringBuilder(s);
+ for (String ss : other) {
+ sb.append(ss);
+ }
+ return sb.toString().getBytes();
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/75d0f49d/hbase-server/src/main/java/org/apache/hadoop/hbase/backup/impl/FullTableBackupClient.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/backup/impl/FullTableBackupClient.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/backup/impl/FullTableBackupClient.java
new file mode 100644
index 0000000..77d1184
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/backup/impl/FullTableBackupClient.java
@@ -0,0 +1,189 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.backup.impl;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.backup.BackupCopyJob;
+import org.apache.hadoop.hbase.backup.BackupInfo;
+import org.apache.hadoop.hbase.backup.BackupInfo.BackupPhase;
+import org.apache.hadoop.hbase.backup.BackupInfo.BackupState;
+import org.apache.hadoop.hbase.backup.BackupRequest;
+import org.apache.hadoop.hbase.backup.BackupRestoreFactory;
+import org.apache.hadoop.hbase.backup.BackupType;
+import org.apache.hadoop.hbase.backup.master.LogRollMasterProcedureManager;
+import org.apache.hadoop.hbase.backup.util.BackupUtils;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+
+/**
+ * Full table backup implementation
+ *
+ */
+@InterfaceAudience.Private
+public class FullTableBackupClient extends TableBackupClient {
+ private static final Log LOG = LogFactory.getLog(FullTableBackupClient.class);
+
+ public FullTableBackupClient(final Connection conn, final String backupId, BackupRequest request)
+ throws IOException {
+ super(conn, backupId, request);
+ }
+
+ /**
+ * Do snapshot copy.
+ * @param backupInfo backup info
+ * @throws Exception exception
+ */
+ private void snapshotCopy(BackupInfo backupInfo) throws Exception {
+ LOG.info("Snapshot copy is starting.");
+
+ // set overall backup phase: snapshot_copy
+ backupInfo.setPhase(BackupPhase.SNAPSHOTCOPY);
+
+ // call ExportSnapshot to copy files based on hbase snapshot for backup
+ // ExportSnapshot only support single snapshot export, need loop for multiple tables case
+ BackupCopyJob copyService = BackupRestoreFactory.getBackupCopyJob(conf);
+
+ // number of snapshots matches number of tables
+ float numOfSnapshots = backupInfo.getSnapshotNames().size();
+
+ LOG.debug("There are " + (int) numOfSnapshots + " snapshots to be copied.");
+
+ for (TableName table : backupInfo.getTables()) {
+ // Currently we simply set the sub copy tasks by counting the table snapshot number, we can
+ // calculate the real files' size for the percentage in the future.
+ // backupCopier.setSubTaskPercntgInWholeTask(1f / numOfSnapshots);
+ int res = 0;
+ String[] args = new String[4];
+ args[0] = "-snapshot";
+ args[1] = backupInfo.getSnapshotName(table);
+ args[2] = "-copy-to";
+ args[3] = backupInfo.getTableBackupDir(table);
+
+ LOG.debug("Copy snapshot " + args[1] + " to " + args[3]);
+ res = copyService.copy(backupInfo, backupManager, conf, BackupType.FULL, args);
+ // if one snapshot export failed, do not continue for remained snapshots
+ if (res != 0) {
+ LOG.error("Exporting Snapshot " + args[1] + " failed with return code: " + res + ".");
+
+ throw new IOException("Failed of exporting snapshot " + args[1] + " to " + args[3]
+ + " with reason code " + res);
+ }
+ LOG.info("Snapshot copy " + args[1] + " finished.");
+ }
+ }
+
+ /**
+ * Backup request execution
+ * @throws IOException
+ */
+ @Override
+ public void execute() throws IOException {
+
+ try (Admin admin = conn.getAdmin();) {
+
+ // Begin BACKUP
+ beginBackup(backupManager, backupInfo);
+ String savedStartCode = null;
+ boolean firstBackup = false;
+ // do snapshot for full table backup
+
+ savedStartCode = backupManager.readBackupStartCode();
+ firstBackup = savedStartCode == null || Long.parseLong(savedStartCode) == 0L;
+ if (firstBackup) {
+ // This is our first backup. Let's put some marker to system table so that we can hold the logs
+ // while we do the backup.
+ backupManager.writeBackupStartCode(0L);
+ }
+ // We roll log here before we do the snapshot. It is possible there is duplicate data
+ // in the log that is already in the snapshot. But if we do it after the snapshot, we
+ // could have data loss.
+ // A better approach is to do the roll log on each RS in the same global procedure as
+ // the snapshot.
+ LOG.info("Execute roll log procedure for full backup ...");
+
+ Map<String, String> props = new HashMap<String, String>();
+ props.put("backupRoot", backupInfo.getBackupRootDir());
+ admin.execProcedure(LogRollMasterProcedureManager.ROLLLOG_PROCEDURE_SIGNATURE,
+ LogRollMasterProcedureManager.ROLLLOG_PROCEDURE_NAME, props);
+
+ newTimestamps = backupManager.readRegionServerLastLogRollResult();
+ if (firstBackup) {
+ // Updates registered log files
+ // We record ALL old WAL files as registered, because
+ // this is a first full backup in the system and these
+ // files are not needed for next incremental backup
+ List<String> logFiles = BackupUtils.getWALFilesOlderThan(conf, newTimestamps);
+ backupManager.recordWALFiles(logFiles);
+ }
+
+ // SNAPSHOT_TABLES:
+ backupInfo.setPhase(BackupPhase.SNAPSHOT);
+ for (TableName tableName : tableList) {
+ String snapshotName =
+ "snapshot_" + Long.toString(EnvironmentEdgeManager.currentTime()) + "_"
+ + tableName.getNamespaceAsString() + "_" + tableName.getQualifierAsString();
+
+ admin.snapshot(snapshotName, tableName);
+
+ backupInfo.setSnapshotName(tableName, snapshotName);
+ }
+
+ // SNAPSHOT_COPY:
+ // do snapshot copy
+ LOG.debug("snapshot copy for " + backupId);
+ snapshotCopy(backupInfo);
+ // Updates incremental backup table set
+ backupManager.addIncrementalBackupTableSet(backupInfo.getTables());
+
+ // BACKUP_COMPLETE:
+ // set overall backup status: complete. Here we make sure to complete the backup.
+ // After this checkpoint, even if entering cancel process, will let the backup finished
+ backupInfo.setState(BackupState.COMPLETE);
+ // The table list in backupInfo is good for both full backup and incremental backup.
+ // For incremental backup, it contains the incremental backup table set.
+ backupManager.writeRegionServerLogTimestamp(backupInfo.getTables(), newTimestamps);
+
+ HashMap<TableName, HashMap<String, Long>> newTableSetTimestampMap =
+ backupManager.readLogTimestampMap();
+
+ Long newStartCode =
+ BackupUtils.getMinValue(BackupUtils
+ .getRSLogTimestampMins(newTableSetTimestampMap));
+ backupManager.writeBackupStartCode(newStartCode);
+
+ // backup complete
+ completeBackup(conn, backupInfo, backupManager, BackupType.FULL, conf);
+ } catch (Exception e) {
+ failBackup(conn, backupInfo, backupManager, e, "Unexpected BackupException : ",
+ BackupType.FULL, conf);
+ throw new IOException(e);
+ }
+
+ }
+
+}