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.impl;
019
020import static org.apache.hadoop.hbase.HConstants.REPLICATION_BULKLOAD_ENABLE_KEY;
021import static org.apache.hadoop.hbase.HConstants.REPLICATION_SCOPE_GLOBAL;
022import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.BACKUP_ATTEMPTS_PAUSE_MS_KEY;
023import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.BACKUP_MAX_ATTEMPTS_KEY;
024import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.CONF_CONTINUOUS_BACKUP_WAL_DIR;
025import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.CONTINUOUS_BACKUP_OFFSET_UPDATE_INTERVAL_MS;
026import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.CONTINUOUS_BACKUP_OFFSET_UPDATE_SIZE_THRESHOLD;
027import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.CONTINUOUS_BACKUP_REPLICATION_PEER;
028import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.DEFAULT_BACKUP_ATTEMPTS_PAUSE_MS;
029import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.DEFAULT_BACKUP_MAX_ATTEMPTS;
030import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.DEFAULT_CONTINUOUS_BACKUP_OFFSET_UPDATE_INTERVAL_MS;
031import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.DEFAULT_CONTINUOUS_BACKUP_OFFSET_UPDATE_SIZE_THRESHOLD;
032import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.DEFAULT_CONTINUOUS_BACKUP_REPLICATION_ENDPOINT;
033import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.JOB_NAME_CONF_KEY;
034import static org.apache.hadoop.hbase.backup.replication.ContinuousBackupReplicationEndpoint.CONF_BACKUP_ROOT_DIR;
035import static org.apache.hadoop.hbase.backup.replication.ContinuousBackupReplicationEndpoint.CONF_PEER_UUID;
036import static org.apache.hadoop.hbase.replication.ReplicationUtils.OFFSET_UPDATE_INTERVAL_MS_KEY;
037import static org.apache.hadoop.hbase.replication.ReplicationUtils.OFFSET_UPDATE_SIZE_THRESHOLD_KEY;
038
039import java.io.IOException;
040import java.util.ArrayList;
041import java.util.HashMap;
042import java.util.List;
043import java.util.Map;
044import java.util.UUID;
045import java.util.stream.Collectors;
046import org.apache.hadoop.hbase.TableName;
047import org.apache.hadoop.hbase.backup.BackupCopyJob;
048import org.apache.hadoop.hbase.backup.BackupInfo;
049import org.apache.hadoop.hbase.backup.BackupInfo.BackupPhase;
050import org.apache.hadoop.hbase.backup.BackupInfo.BackupState;
051import org.apache.hadoop.hbase.backup.BackupRequest;
052import org.apache.hadoop.hbase.backup.BackupRestoreFactory;
053import org.apache.hadoop.hbase.backup.BackupType;
054import org.apache.hadoop.hbase.backup.util.BackupUtils;
055import org.apache.hadoop.hbase.client.Admin;
056import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
057import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
058import org.apache.hadoop.hbase.client.Connection;
059import org.apache.hadoop.hbase.client.TableDescriptor;
060import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
061import org.apache.hadoop.hbase.replication.ReplicationException;
062import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
063import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
064import org.apache.yetus.audience.InterfaceAudience;
065import org.slf4j.Logger;
066import org.slf4j.LoggerFactory;
067
068/**
069 * Full table backup implementation
070 */
071@InterfaceAudience.Private
072public class FullTableBackupClient extends TableBackupClient {
073  private static final Logger LOG = LoggerFactory.getLogger(FullTableBackupClient.class);
074
075  public FullTableBackupClient() {
076  }
077
078  public FullTableBackupClient(final Connection conn, final String backupId, BackupRequest request)
079    throws IOException {
080    super(conn, backupId, request);
081  }
082
083  /**
084   * Do snapshot copy.
085   * @param backupInfo backup info
086   * @throws IOException exception
087   */
088  protected void snapshotCopy(BackupInfo backupInfo) throws IOException {
089    LOG.info("Snapshot copy is starting.");
090
091    // set overall backup phase: snapshot_copy
092    backupInfo.setPhase(BackupPhase.SNAPSHOTCOPY);
093
094    // call ExportSnapshot to copy files based on hbase snapshot for backup
095    // ExportSnapshot only support single snapshot export, need loop for multiple tables case
096    BackupCopyJob copyService = BackupRestoreFactory.getBackupCopyJob(conf);
097
098    // number of snapshots matches number of tables
099    float numOfSnapshots = backupInfo.getSnapshotNames().size();
100
101    LOG.debug("There are " + (int) numOfSnapshots + " snapshots to be copied.");
102
103    for (TableName table : backupInfo.getTables()) {
104      // Currently we simply set the sub copy tasks by counting the table snapshot number, we can
105      // calculate the real files' size for the percentage in the future.
106      // backupCopier.setSubTaskPercntgInWholeTask(1f / numOfSnapshots);
107      int res;
108      ArrayList<String> argsList = new ArrayList<>();
109      argsList.add("-snapshot");
110      argsList.add(backupInfo.getSnapshotName(table));
111      argsList.add("-copy-to");
112      argsList.add(backupInfo.getTableBackupDir(table));
113      if (backupInfo.getBandwidth() > -1) {
114        argsList.add("-bandwidth");
115        argsList.add(String.valueOf(backupInfo.getBandwidth()));
116      }
117      if (backupInfo.getWorkers() > -1) {
118        argsList.add("-mappers");
119        argsList.add(String.valueOf(backupInfo.getWorkers()));
120      }
121      if (backupInfo.getNoChecksumVerify()) {
122        argsList.add("-no-checksum-verify");
123      }
124
125      String[] args = argsList.toArray(new String[0]);
126
127      String jobname = "Full-Backup_" + backupInfo.getBackupId() + "_" + table.getNameAsString();
128      if (LOG.isDebugEnabled()) {
129        LOG.debug("Setting snapshot copy job name to : " + jobname);
130      }
131      conf.set(JOB_NAME_CONF_KEY, jobname);
132
133      LOG.debug("Copy snapshot " + args[1] + " to " + args[3]);
134      res = copyService.copy(backupInfo, backupManager, conf, BackupType.FULL, args);
135
136      // if one snapshot export failed, do not continue for remained snapshots
137      if (res != 0) {
138        LOG.error("Exporting Snapshot " + args[1] + " failed with return code: " + res + ".");
139
140        throw new IOException("Failed of exporting snapshot " + args[1] + " to " + args[3]
141          + " with reason code " + res);
142      }
143
144      conf.unset(JOB_NAME_CONF_KEY);
145      LOG.info("Snapshot copy " + args[1] + " finished.");
146    }
147  }
148
149  /**
150   * Backup request execution.
151   * @throws IOException if the execution of the backup fails
152   */
153  @Override
154  public void execute() throws IOException {
155    try (Admin admin = conn.getAdmin()) {
156      beginBackup(backupManager, backupInfo);
157
158      // Gather the bulk loads being tracked by the system, which can be deleted (since their data
159      // will be part of the snapshot being taken). We gather this list before taking the actual
160      // snapshots for the same reason as the log rolls.
161      List<BulkLoad> bulkLoadsToDelete = backupManager.readBulkloadRows(tableList);
162
163      if (backupInfo.isContinuousBackupEnabled()) {
164        handleContinuousBackup(admin);
165      } else {
166        handleNonContinuousBackup(admin);
167      }
168
169      backupManager
170        .deleteBulkLoadedRows(bulkLoadsToDelete.stream().map(BulkLoad::getRowKey).toList());
171
172      completeBackup(conn, backupInfo, BackupType.FULL, conf);
173    } catch (Exception e) {
174      failBackup(conn, backupInfo, backupManager, e, "Unexpected BackupException : ",
175        BackupType.FULL, conf);
176      throw new IOException(e);
177    }
178  }
179
180  private void handleContinuousBackup(Admin admin) throws IOException {
181    backupInfo.setPhase(BackupInfo.BackupPhase.SETUP_WAL_REPLICATION);
182    long startTimestamp = startContinuousWALBackup(admin);
183    backupManager.addContinuousBackupTableSet(backupInfo.getTables(), startTimestamp);
184
185    // Updating the start time of this backup to reflect the actual beginning of the full backup.
186    // So far, we have only set up continuous WAL replication, but the full backup has not yet
187    // started.
188    // Setting the correct start time is crucial for Point-In-Time Recovery (PITR).
189    // When selecting a backup for PITR, we must ensure that the backup started **on or after** the
190    // starting time of the WALs. If WAL streaming began later, we couldn't guarantee that WALs
191    // exist for the entire period between the backup's start time and the desired PITR timestamp.
192    backupInfo.setStartTs(startTimestamp);
193
194    performBackupSnapshots(admin);
195
196    // set overall backup status: complete. Here we make sure to complete the backup.
197    // After this checkpoint, even if entering cancel process, will let the backup finished
198    backupInfo.setState(BackupState.COMPLETE);
199
200    if (!conf.getBoolean(REPLICATION_BULKLOAD_ENABLE_KEY, false)) {
201      System.out.println("WARNING: Bulkload replication is not enabled. "
202        + "Since continuous backup is using HBase replication, bulk loaded files won't be backed up as part of continuous backup. "
203        + "To ensure bulk-loaded files are backed up, enable bulkload replication "
204        + "(hbase.replication.bulkload.enabled=true) and configure a unique cluster ID using "
205        + "hbase.replication.cluster.id. This cluster ID is required by the replication framework "
206        + "to uniquely identify clusters, even if continuous backup itself does not directly rely on it.");
207    }
208  }
209
210  private void handleNonContinuousBackup(Admin admin) throws IOException {
211    performLogRoll();
212    performBackupSnapshots(admin);
213    backupManager.addIncrementalBackupTableSet(backupInfo.getTables());
214
215    // set overall backup status: complete. Here we make sure to complete the backup.
216    // After this checkpoint, even if entering cancel process, will let the backup finished
217    backupInfo.setState(BackupState.COMPLETE);
218
219    updateBackupMetadata();
220  }
221
222  private void performLogRoll() throws IOException {
223    // We roll log here before we do the snapshot. It is possible there is duplicate data
224    // in the log that is already in the snapshot. But if we do it after the snapshot, we
225    // could have data loss.
226    // A better approach is to do the roll log on each RS in the same global procedure as
227    // the snapshot.
228    LOG.info("Execute roll log procedure for full backup ...");
229    BackupUtils.logRoll(conn, backupInfo.getBackupRootDir(), conf);
230    newTimestamps = backupManager.readRegionServerLastLogRollResult();
231  }
232
233  private void performBackupSnapshots(Admin admin) throws IOException {
234    backupInfo.setPhase(BackupPhase.SNAPSHOT);
235    performSnapshots(admin);
236    LOG.debug("Performing snapshot copy for backup ID: {}", backupInfo.getBackupId());
237    snapshotCopy(backupInfo);
238  }
239
240  private void performSnapshots(Admin admin) throws IOException {
241    backupInfo.setPhase(BackupPhase.SNAPSHOT);
242
243    for (TableName tableName : tableList) {
244      String snapshotName = String.format("snapshot_%d_%s_%s", EnvironmentEdgeManager.currentTime(),
245        tableName.getNamespaceAsString(), tableName.getQualifierAsString());
246      snapshotTable(admin, tableName, snapshotName);
247      backupInfo.setSnapshotName(tableName, snapshotName);
248    }
249  }
250
251  private void updateBackupMetadata() throws IOException {
252    // The table list in backupInfo is good for both full backup and incremental backup.
253    // For incremental backup, it contains the incremental backup table set.
254    backupManager.writeRegionServerLogTimestamp(backupInfo.getTables(), newTimestamps);
255    Map<TableName, Map<String, Long>> timestampMap = backupManager.readLogTimestampMap();
256    backupInfo.setTableSetTimestampMap(timestampMap);
257  }
258
259  private long startContinuousWALBackup(Admin admin) throws IOException {
260    enableTableReplication(admin);
261    if (continuousBackupReplicationPeerExists(admin)) {
262      updateContinuousBackupReplicationPeer(admin);
263    } else {
264      addContinuousBackupReplicationPeer(admin);
265    }
266    LOG.info("Continuous WAL Backup setup completed.");
267    return EnvironmentEdgeManager.getDelegate().currentTime();
268  }
269
270  private void enableTableReplication(Admin admin) throws IOException {
271    for (TableName table : tableList) {
272      TableDescriptor tableDescriptor = admin.getDescriptor(table);
273      TableDescriptorBuilder tableDescriptorBuilder =
274        TableDescriptorBuilder.newBuilder(tableDescriptor);
275
276      for (ColumnFamilyDescriptor cfDescriptor : tableDescriptor.getColumnFamilies()) {
277        if (cfDescriptor.getScope() != REPLICATION_SCOPE_GLOBAL) {
278          ColumnFamilyDescriptor newCfDescriptor = ColumnFamilyDescriptorBuilder
279            .newBuilder(cfDescriptor).setScope(REPLICATION_SCOPE_GLOBAL).build();
280
281          tableDescriptorBuilder.modifyColumnFamily(newCfDescriptor);
282        }
283      }
284
285      admin.modifyTable(tableDescriptorBuilder.build());
286      LOG.info("Enabled Global replication scope for table: {}", table);
287    }
288  }
289
290  private void updateContinuousBackupReplicationPeer(Admin admin) throws IOException {
291    Map<TableName, List<String>> tableMap = tableList.stream()
292      .collect(Collectors.toMap(tableName -> tableName, tableName -> new ArrayList<>()));
293
294    try {
295      if (!admin.isReplicationPeerEnabled(CONTINUOUS_BACKUP_REPLICATION_PEER)) {
296        admin.enableReplicationPeer(CONTINUOUS_BACKUP_REPLICATION_PEER);
297      }
298      admin.appendReplicationPeerTableCFs(CONTINUOUS_BACKUP_REPLICATION_PEER, tableMap);
299      LOG.info("Updated replication peer {} with table and column family map.",
300        CONTINUOUS_BACKUP_REPLICATION_PEER);
301    } catch (ReplicationException e) {
302      LOG.error("Error while updating the replication peer: {}. Error: {}",
303        CONTINUOUS_BACKUP_REPLICATION_PEER, e.getMessage(), e);
304      throw new IOException("Error while updating the continuous backup replication peer.", e);
305    }
306  }
307
308  private void addContinuousBackupReplicationPeer(Admin admin) throws IOException {
309    String backupWalDir = conf.get(CONF_CONTINUOUS_BACKUP_WAL_DIR);
310
311    if (backupWalDir == null || backupWalDir.isEmpty()) {
312      String errorMsg = "WAL Directory is not specified for continuous backup.";
313      LOG.error(errorMsg);
314      throw new IOException(errorMsg);
315    }
316
317    Map<String, String> additionalArgs = new HashMap<>();
318    additionalArgs.put(CONF_PEER_UUID, UUID.randomUUID().toString());
319    additionalArgs.put(CONF_BACKUP_ROOT_DIR, backupWalDir);
320    additionalArgs.put(OFFSET_UPDATE_INTERVAL_MS_KEY,
321      String.valueOf(conf.getLong(CONTINUOUS_BACKUP_OFFSET_UPDATE_INTERVAL_MS,
322        DEFAULT_CONTINUOUS_BACKUP_OFFSET_UPDATE_INTERVAL_MS)));
323
324    additionalArgs.put(OFFSET_UPDATE_SIZE_THRESHOLD_KEY,
325      String.valueOf(conf.getLong(CONTINUOUS_BACKUP_OFFSET_UPDATE_SIZE_THRESHOLD,
326        DEFAULT_CONTINUOUS_BACKUP_OFFSET_UPDATE_SIZE_THRESHOLD)));
327
328    Map<TableName, List<String>> tableMap = tableList.stream()
329      .collect(Collectors.toMap(tableName -> tableName, tableName -> new ArrayList<>()));
330
331    ReplicationPeerConfig peerConfig = ReplicationPeerConfig.newBuilder()
332      .setReplicationEndpointImpl(DEFAULT_CONTINUOUS_BACKUP_REPLICATION_ENDPOINT)
333      .setReplicateAllUserTables(false).setTableCFsMap(tableMap).putAllConfiguration(additionalArgs)
334      .build();
335
336    try {
337      admin.addReplicationPeer(CONTINUOUS_BACKUP_REPLICATION_PEER, peerConfig, true);
338      LOG.info("Successfully added replication peer with ID: {}",
339        CONTINUOUS_BACKUP_REPLICATION_PEER);
340    } catch (IOException e) {
341      LOG.error("Failed to add replication peer with ID: {}. Error: {}",
342        CONTINUOUS_BACKUP_REPLICATION_PEER, e.getMessage(), e);
343      throw e;
344    }
345  }
346
347  private boolean continuousBackupReplicationPeerExists(Admin admin) throws IOException {
348    return admin.listReplicationPeers().stream()
349      .anyMatch(peer -> peer.getPeerId().equals(CONTINUOUS_BACKUP_REPLICATION_PEER));
350  }
351
352  protected void snapshotTable(Admin admin, TableName tableName, String snapshotName)
353    throws IOException {
354    int maxAttempts = conf.getInt(BACKUP_MAX_ATTEMPTS_KEY, DEFAULT_BACKUP_MAX_ATTEMPTS);
355    int pause = conf.getInt(BACKUP_ATTEMPTS_PAUSE_MS_KEY, DEFAULT_BACKUP_ATTEMPTS_PAUSE_MS);
356    int attempts = 0;
357
358    while (attempts++ < maxAttempts) {
359      try {
360        admin.snapshot(snapshotName, tableName);
361        return;
362      } catch (IOException ee) {
363        LOG.warn("Snapshot attempt " + attempts + " failed for table " + tableName
364          + ", sleeping for " + pause + "ms", ee);
365        if (attempts < maxAttempts) {
366          try {
367            Thread.sleep(pause);
368          } catch (InterruptedException e) {
369            Thread.currentThread().interrupt();
370            break;
371          }
372        }
373      }
374    }
375    throw new IOException("Failed to snapshot table " + tableName);
376  }
377}