001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.backup; 019 020import java.io.FileNotFoundException; 021import java.io.IOException; 022import java.io.InterruptedIOException; 023import java.util.ArrayList; 024import java.util.Collection; 025import java.util.Collections; 026import java.util.HashMap; 027import java.util.List; 028import java.util.Map; 029import java.util.Queue; 030import java.util.concurrent.ConcurrentLinkedQueue; 031import java.util.concurrent.ExecutionException; 032import java.util.concurrent.Future; 033import java.util.concurrent.ThreadFactory; 034import java.util.concurrent.ThreadPoolExecutor; 035import java.util.concurrent.TimeUnit; 036import java.util.concurrent.atomic.AtomicInteger; 037import java.util.function.Function; 038import java.util.stream.Collectors; 039import java.util.stream.Stream; 040import org.apache.hadoop.conf.Configuration; 041import org.apache.hadoop.fs.FileStatus; 042import org.apache.hadoop.fs.FileSystem; 043import org.apache.hadoop.fs.Path; 044import org.apache.hadoop.fs.PathFilter; 045import org.apache.hadoop.hbase.HConstants; 046import org.apache.hadoop.hbase.client.RegionInfo; 047import org.apache.hadoop.hbase.regionserver.HStoreFile; 048import org.apache.hadoop.hbase.util.Bytes; 049import org.apache.hadoop.hbase.util.CommonFSUtils; 050import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 051import org.apache.hadoop.hbase.util.FSUtils; 052import org.apache.hadoop.hbase.util.HFileArchiveUtil; 053import org.apache.hadoop.hbase.util.Threads; 054import org.apache.hadoop.io.MultipleIOException; 055import org.apache.yetus.audience.InterfaceAudience; 056import org.slf4j.Logger; 057import org.slf4j.LoggerFactory; 058 059import org.apache.hbase.thirdparty.com.google.common.base.Preconditions; 060 061/** 062 * Utility class to handle the removal of HFiles (or the respective {@link HStoreFile StoreFiles}) 063 * for a HRegion from the {@link FileSystem}. The hfiles will be archived or deleted, depending on 064 * the state of the system. 065 */ 066@InterfaceAudience.Private 067public class HFileArchiver { 068 private static final Logger LOG = LoggerFactory.getLogger(HFileArchiver.class); 069 private static final String SEPARATOR = "."; 070 071 /** Number of retries in case of fs operation failure */ 072 private static final int DEFAULT_RETRIES_NUMBER = 3; 073 074 private static final Function<File, Path> FUNC_FILE_TO_PATH = new Function<File, Path>() { 075 @Override 076 public Path apply(File file) { 077 return file == null ? null : file.getPath(); 078 } 079 }; 080 081 private static ThreadPoolExecutor archiveExecutor; 082 083 private HFileArchiver() { 084 // hidden ctor since this is just a util 085 } 086 087 /** Returns True if the Region exits in the filesystem. */ 088 public static boolean exists(Configuration conf, FileSystem fs, RegionInfo info) 089 throws IOException { 090 Path rootDir = CommonFSUtils.getRootDir(conf); 091 Path regionDir = FSUtils.getRegionDirFromRootDir(rootDir, info); 092 return fs.exists(regionDir); 093 } 094 095 /** 096 * Cleans up all the files for a HRegion by archiving the HFiles to the archive directory 097 * @param conf the configuration to use 098 * @param fs the file system object 099 * @param info RegionInfo for region to be deleted 100 */ 101 public static void archiveRegion(Configuration conf, FileSystem fs, RegionInfo info) 102 throws IOException { 103 Path rootDir = CommonFSUtils.getRootDir(conf); 104 archiveRegion(conf, fs, rootDir, CommonFSUtils.getTableDir(rootDir, info.getTable()), 105 FSUtils.getRegionDirFromRootDir(rootDir, info)); 106 } 107 108 /** 109 * Remove an entire region from the table directory via archiving the region's hfiles. 110 * @param fs {@link FileSystem} from which to remove the region 111 * @param rootdir {@link Path} to the root directory where hbase files are stored (for building 112 * the archive path) 113 * @param tableDir {@link Path} to where the table is being stored (for building the archive 114 * path) 115 * @param regionDir {@link Path} to where a region is being stored (for building the archive path) 116 * @return <tt>true</tt> if the region was successfully deleted. <tt>false</tt> if the filesystem 117 * operations could not complete. 118 * @throws IOException if the request cannot be completed 119 */ 120 public static boolean archiveRegion(Configuration conf, FileSystem fs, Path rootdir, 121 Path tableDir, Path regionDir) throws IOException { 122 // otherwise, we archive the files 123 // make sure we can archive 124 if (tableDir == null || regionDir == null) { 125 LOG.error("No archive directory could be found because tabledir (" + tableDir 126 + ") or regiondir (" + regionDir + "was null. Deleting files instead."); 127 if (regionDir != null) { 128 deleteRegionWithoutArchiving(fs, regionDir); 129 } 130 // we should have archived, but failed to. Doesn't matter if we deleted 131 // the archived files correctly or not. 132 return false; 133 } 134 135 LOG.debug("ARCHIVING {}", regionDir); 136 137 // make sure the regiondir lives under the tabledir 138 Preconditions.checkArgument(regionDir.toString().startsWith(tableDir.toString())); 139 Path regionArchiveDir = HFileArchiveUtil.getRegionArchiveDir(rootdir, 140 CommonFSUtils.getTableName(tableDir), regionDir.getName()); 141 142 FileStatusConverter getAsFile = new FileStatusConverter(fs); 143 // otherwise, we attempt to archive the store files 144 145 // build collection of just the store directories to archive 146 Collection<File> toArchive = new ArrayList<>(); 147 final PathFilter dirFilter = new FSUtils.DirFilter(fs); 148 PathFilter nonHidden = new PathFilter() { 149 @Override 150 public boolean accept(Path file) { 151 return dirFilter.accept(file) && !file.getName().startsWith("."); 152 } 153 }; 154 FileStatus[] storeDirs = CommonFSUtils.listStatus(fs, regionDir, nonHidden); 155 // if there no files, we can just delete the directory and return; 156 if (storeDirs == null) { 157 LOG.debug("Directory {} empty.", regionDir); 158 return deleteRegionWithoutArchiving(fs, regionDir); 159 } 160 161 // convert the files in the region to a File 162 Stream.of(storeDirs).map(getAsFile).forEachOrdered(toArchive::add); 163 LOG.debug("Archiving " + toArchive); 164 List<File> failedArchive = resolveAndArchive(conf, fs, regionArchiveDir, toArchive, 165 EnvironmentEdgeManager.currentTime()); 166 if (!failedArchive.isEmpty()) { 167 throw new FailedArchiveException( 168 "Failed to archive/delete all the files for region:" + regionDir.getName() + " into " 169 + regionArchiveDir + ". Something is probably awry on the filesystem.", 170 failedArchive.stream().map(FUNC_FILE_TO_PATH).collect(Collectors.toList())); 171 } 172 // if that was successful, then we delete the region 173 return deleteRegionWithoutArchiving(fs, regionDir); 174 } 175 176 /** 177 * Archive the specified regions in parallel. 178 * @param conf the configuration to use 179 * @param fs {@link FileSystem} from which to remove the region 180 * @param rootDir {@link Path} to the root directory where hbase files are stored (for 181 * building the archive path) 182 * @param tableDir {@link Path} to where the table is being stored (for building the archive 183 * path) 184 * @param regionDirList {@link Path} to where regions are being stored (for building the archive 185 * path) 186 * @throws IOException if the request cannot be completed 187 */ 188 public static void archiveRegions(Configuration conf, FileSystem fs, Path rootDir, Path tableDir, 189 List<Path> regionDirList) throws IOException { 190 List<Future<Void>> futures = new ArrayList<>(regionDirList.size()); 191 for (Path regionDir : regionDirList) { 192 Future<Void> future = getArchiveExecutor(conf).submit(() -> { 193 archiveRegion(conf, fs, rootDir, tableDir, regionDir); 194 return null; 195 }); 196 futures.add(future); 197 } 198 try { 199 for (Future<Void> future : futures) { 200 future.get(); 201 } 202 } catch (InterruptedException e) { 203 throw new InterruptedIOException(e.getMessage()); 204 } catch (ExecutionException e) { 205 throw new IOException(e.getCause()); 206 } 207 } 208 209 private static synchronized ThreadPoolExecutor getArchiveExecutor(final Configuration conf) { 210 if (archiveExecutor == null) { 211 int maxThreads = conf.getInt("hbase.hfilearchiver.thread.pool.max", 8); 212 archiveExecutor = Threads.getBoundedCachedThreadPool(maxThreads, 30L, TimeUnit.SECONDS, 213 getThreadFactory("HFileArchiver")); 214 215 // Shutdown this ThreadPool in a shutdown hook 216 Runtime.getRuntime().addShutdownHook(new Thread(() -> archiveExecutor.shutdown())); 217 } 218 return archiveExecutor; 219 } 220 221 // We need this method instead of Threads.getNamedThreadFactory() to pass some tests. 222 // The difference from Threads.getNamedThreadFactory() is that it doesn't fix ThreadGroup for 223 // new threads. If we use Threads.getNamedThreadFactory(), we will face ThreadGroup related 224 // issues in some tests. 225 private static ThreadFactory getThreadFactory(String archiverName) { 226 return new ThreadFactory() { 227 final AtomicInteger threadNumber = new AtomicInteger(1); 228 229 @Override 230 public Thread newThread(Runnable r) { 231 final String name = archiverName + "-" + threadNumber.getAndIncrement(); 232 Thread t = new Thread(r, name); 233 t.setDaemon(true); 234 return t; 235 } 236 }; 237 } 238 239 /** 240 * Remove from the specified region the store files of the specified column family, either by 241 * archiving them or outright deletion 242 * @param fs the filesystem where the store files live 243 * @param conf {@link Configuration} to examine to determine the archive directory 244 * @param parent Parent region hosting the store files 245 * @param tableDir {@link Path} to where the table is being stored (for building the archive path) 246 * @param family the family hosting the store files 247 * @throws IOException if the files could not be correctly disposed. 248 */ 249 public static void archiveFamily(FileSystem fs, Configuration conf, RegionInfo parent, 250 Path tableDir, byte[] family) throws IOException { 251 Path familyDir = new Path(tableDir, new Path(parent.getEncodedName(), Bytes.toString(family))); 252 archiveFamilyByFamilyDir(fs, conf, parent, familyDir, family); 253 } 254 255 /** 256 * Removes from the specified region the store files of the specified column family, either by 257 * archiving them or outright deletion 258 * @param fs the filesystem where the store files live 259 * @param conf {@link Configuration} to examine to determine the archive directory 260 * @param parent Parent region hosting the store files 261 * @param familyDir {@link Path} to where the family is being stored 262 * @param family the family hosting the store files 263 * @throws IOException if the files could not be correctly disposed. 264 */ 265 public static void archiveFamilyByFamilyDir(FileSystem fs, Configuration conf, RegionInfo parent, 266 Path familyDir, byte[] family) throws IOException { 267 FileStatus[] storeFiles = CommonFSUtils.listStatus(fs, familyDir); 268 if (storeFiles == null) { 269 LOG.debug("No files to dispose of in {}, family={}", parent.getRegionNameAsString(), 270 Bytes.toString(family)); 271 return; 272 } 273 274 FileStatusConverter getAsFile = new FileStatusConverter(fs); 275 Collection<File> toArchive = Stream.of(storeFiles).map(getAsFile).collect(Collectors.toList()); 276 Path storeArchiveDir = HFileArchiveUtil.getStoreArchivePath(conf, parent, family); 277 278 // do the actual archive 279 List<File> failedArchive = 280 resolveAndArchive(conf, fs, storeArchiveDir, toArchive, EnvironmentEdgeManager.currentTime()); 281 if (!failedArchive.isEmpty()) { 282 throw new FailedArchiveException( 283 "Failed to archive/delete all the files for region:" 284 + Bytes.toString(parent.getRegionName()) + ", family:" + Bytes.toString(family) + " into " 285 + storeArchiveDir + ". Something is probably awry on the filesystem.", 286 failedArchive.stream().map(FUNC_FILE_TO_PATH).collect(Collectors.toList())); 287 } 288 } 289 290 /** 291 * Remove the store files, either by archiving them or outright deletion 292 * @param conf {@link Configuration} to examine to determine the archive directory 293 * @param fs the filesystem where the store files live 294 * @param regionInfo {@link RegionInfo} of the region hosting the store files 295 * @param family the family hosting the store files 296 * @param compactedFiles files to be disposed of. No further reading of these files should be 297 * attempted; otherwise likely to cause an {@link IOException} 298 * @throws IOException if the files could not be correctly disposed. 299 */ 300 public static void archiveStoreFiles(Configuration conf, FileSystem fs, RegionInfo regionInfo, 301 Path tableDir, byte[] family, Collection<HStoreFile> compactedFiles) throws IOException { 302 Path storeArchiveDir = HFileArchiveUtil.getStoreArchivePath(conf, regionInfo, tableDir, family); 303 archive(conf, fs, regionInfo, family, compactedFiles, storeArchiveDir); 304 } 305 306 /** 307 * Archive recovered edits using existing logic for archiving store files. This is currently only 308 * relevant when <b>hbase.region.archive.recovered.edits</b> is true, as recovered edits shouldn't 309 * be kept after replay. In theory, we could use very same method available for archiving store 310 * files, but supporting WAL dir and store files on different FileSystems added the need for extra 311 * validation of the passed FileSystem instance and the path where the archiving edits should be 312 * placed. 313 * @param conf {@link Configuration} to determine the archive directory. 314 * @param fs the filesystem used for storing WAL files. 315 * @param regionInfo {@link RegionInfo} a pseudo region representation for the archiving logic. 316 * @param family a pseudo familiy representation for the archiving logic. 317 * @param replayedEdits the recovered edits to be archived. 318 * @throws IOException if files can't be achived due to some internal error. 319 */ 320 public static void archiveRecoveredEdits(Configuration conf, FileSystem fs, RegionInfo regionInfo, 321 byte[] family, Collection<HStoreFile> replayedEdits) throws IOException { 322 String workingDir = conf.get(CommonFSUtils.HBASE_WAL_DIR, conf.get(HConstants.HBASE_DIR)); 323 // extra sanity checks for the right FS 324 Path path = new Path(workingDir); 325 if (path.isAbsoluteAndSchemeAuthorityNull()) { 326 // no schema specified on wal dir value, so it's on same FS as StoreFiles 327 path = new Path(conf.get(HConstants.HBASE_DIR)); 328 } 329 if (path.toUri().getScheme() != null && !path.toUri().getScheme().equals(fs.getScheme())) { 330 throw new IOException( 331 "Wrong file system! Should be " + path.toUri().getScheme() + ", but got " + fs.getScheme()); 332 } 333 path = HFileArchiveUtil.getStoreArchivePathForRootDir(path, regionInfo, family); 334 archive(conf, fs, regionInfo, family, replayedEdits, path); 335 } 336 337 private static void archive(Configuration conf, FileSystem fs, RegionInfo regionInfo, 338 byte[] family, Collection<HStoreFile> compactedFiles, Path storeArchiveDir) throws IOException { 339 // sometimes in testing, we don't have rss, so we need to check for that 340 if (fs == null) { 341 LOG.warn( 342 "Passed filesystem is null, so just deleting files without archiving for {}," + "family={}", 343 Bytes.toString(regionInfo.getRegionName()), Bytes.toString(family)); 344 deleteStoreFilesWithoutArchiving(compactedFiles); 345 return; 346 } 347 348 // short circuit if we don't have any files to delete 349 if (compactedFiles.isEmpty()) { 350 LOG.debug("No files to dispose of, done!"); 351 return; 352 } 353 354 // build the archive path 355 if (regionInfo == null || family == null) 356 throw new IOException("Need to have a region and a family to archive from."); 357 // make sure we don't archive if we can't and that the archive dir exists 358 if (!fs.mkdirs(storeArchiveDir)) { 359 throw new IOException("Could not make archive directory (" + storeArchiveDir + ") for store:" 360 + Bytes.toString(family) + ", deleting compacted files instead."); 361 } 362 363 // otherwise we attempt to archive the store files 364 LOG.debug("Archiving compacted files."); 365 366 // Wrap the storefile into a File 367 StoreToFile getStorePath = new StoreToFile(fs); 368 Collection<File> storeFiles = 369 compactedFiles.stream().map(getStorePath).collect(Collectors.toList()); 370 371 // do the actual archive 372 List<File> failedArchive = resolveAndArchive(conf, fs, storeArchiveDir, storeFiles, 373 EnvironmentEdgeManager.currentTime()); 374 375 if (!failedArchive.isEmpty()) { 376 throw new FailedArchiveException( 377 "Failed to archive/delete all the files for region:" 378 + Bytes.toString(regionInfo.getRegionName()) + ", family:" + Bytes.toString(family) 379 + " into " + storeArchiveDir + ". Something is probably awry on the filesystem.", 380 failedArchive.stream().map(FUNC_FILE_TO_PATH).collect(Collectors.toList())); 381 } 382 } 383 384 /** 385 * Archive the store file 386 * @param fs the filesystem where the store files live 387 * @param regionInfo region hosting the store files 388 * @param conf {@link Configuration} to examine to determine the archive directory 389 * @param tableDir {@link Path} to where the table is being stored (for building the archive 390 * path) 391 * @param family the family hosting the store files 392 * @param storeFile file to be archived 393 * @throws IOException if the files could not be correctly disposed. 394 */ 395 public static void archiveStoreFile(Configuration conf, FileSystem fs, RegionInfo regionInfo, 396 Path tableDir, byte[] family, Path storeFile) throws IOException { 397 Path storeArchiveDir = HFileArchiveUtil.getStoreArchivePath(conf, regionInfo, tableDir, family); 398 // make sure we don't archive if we can't and that the archive dir exists 399 if (!fs.mkdirs(storeArchiveDir)) { 400 throw new IOException("Could not make archive directory (" + storeArchiveDir + ") for store:" 401 + Bytes.toString(family) + ", deleting compacted files instead."); 402 } 403 404 // do the actual archive 405 long start = EnvironmentEdgeManager.currentTime(); 406 File file = new FileablePath(fs, storeFile); 407 if (!resolveAndArchiveFile(storeArchiveDir, file, Long.toString(start))) { 408 throw new IOException("Failed to archive/delete the file for region:" 409 + regionInfo.getRegionNameAsString() + ", family:" + Bytes.toString(family) + " into " 410 + storeArchiveDir + ". Something is probably awry on the filesystem."); 411 } 412 } 413 414 /** 415 * Resolve any conflict with an existing archive file via timestamp-append renaming of the 416 * existing file and then archive the passed in files. 417 * @param fs {@link FileSystem} on which to archive the files 418 * @param baseArchiveDir base archive directory to store the files. If any of the files to archive 419 * are directories, will append the name of the directory to the base 420 * archive directory name, creating a parallel structure. 421 * @param toArchive files/directories that need to be archvied 422 * @param start time the archiving started - used for resolving archive conflicts. 423 * @return the list of failed to archive files. 424 * @throws IOException if an unexpected file operation exception occurred 425 */ 426 private static List<File> resolveAndArchive(Configuration conf, FileSystem fs, 427 Path baseArchiveDir, Collection<File> toArchive, long start) throws IOException { 428 // Early exit if no files to archive 429 if (toArchive.isEmpty()) { 430 LOG.trace("No files to archive, returning an empty list."); 431 return Collections.emptyList(); 432 } 433 434 LOG.trace("Preparing to archive files into directory: {}", baseArchiveDir); 435 436 // Ensure the archive directory exists 437 ensureArchiveDirectoryExists(fs, baseArchiveDir); 438 439 // Thread-safe collection for storing failures 440 Queue<File> failures = new ConcurrentLinkedQueue<>(); 441 String startTime = Long.toString(start); 442 443 // Separate files and directories for processing 444 List<File> filesOnly = new ArrayList<>(); 445 for (File file : toArchive) { 446 if (file.isFile()) { 447 filesOnly.add(file); 448 } else { 449 handleDirectory(conf, fs, baseArchiveDir, failures, file, start); 450 } 451 } 452 453 // Archive files concurrently 454 archiveFilesConcurrently(conf, baseArchiveDir, filesOnly, failures, startTime); 455 456 return new ArrayList<>(failures); // Convert to a List for the return value 457 } 458 459 private static void ensureArchiveDirectoryExists(FileSystem fs, Path baseArchiveDir) 460 throws IOException { 461 if (!fs.exists(baseArchiveDir) && !fs.mkdirs(baseArchiveDir)) { 462 throw new IOException("Failed to create the archive directory: " + baseArchiveDir); 463 } 464 LOG.trace("Archive directory ready: {}", baseArchiveDir); 465 } 466 467 private static void handleDirectory(Configuration conf, FileSystem fs, Path baseArchiveDir, 468 Queue<File> failures, File directory, long start) { 469 LOG.trace("Processing directory: {}, archiving its children.", directory); 470 Path subArchiveDir = new Path(baseArchiveDir, directory.getName()); 471 472 try { 473 Collection<File> children = directory.getChildren(); 474 failures.addAll(resolveAndArchive(conf, fs, subArchiveDir, children, start)); 475 } catch (IOException e) { 476 LOG.warn("Failed to archive directory: {}", directory, e); 477 failures.add(directory); 478 } 479 } 480 481 private static void archiveFilesConcurrently(Configuration conf, Path baseArchiveDir, 482 List<File> files, Queue<File> failures, String startTime) { 483 LOG.trace("Archiving {} files concurrently into directory: {}", files.size(), baseArchiveDir); 484 Map<File, Future<Boolean>> futureMap = new HashMap<>(); 485 // Submit file archiving tasks 486 // default is 16 which comes equal hbase.hstore.blockingStoreFiles default value 487 int maxThreads = conf.getInt("hbase.hfilearchiver.per.region.thread.pool.max", 16); 488 ThreadPoolExecutor hfilesArchiveExecutor = Threads.getBoundedCachedThreadPool(maxThreads, 30L, 489 TimeUnit.SECONDS, getThreadFactory("HFileArchiverPerRegion-")); 490 try { 491 for (File file : files) { 492 Future<Boolean> future = hfilesArchiveExecutor 493 .submit(() -> resolveAndArchiveFile(baseArchiveDir, file, startTime)); 494 futureMap.put(file, future); 495 } 496 497 // Process results of each task 498 for (Map.Entry<File, Future<Boolean>> entry : futureMap.entrySet()) { 499 File file = entry.getKey(); 500 try { 501 if (!entry.getValue().get()) { 502 LOG.warn("Failed to archive file: {} into directory: {}", file, baseArchiveDir); 503 failures.add(file); 504 } 505 } catch (InterruptedException e) { 506 LOG.error("Archiving interrupted for file: {}", file, e); 507 Thread.currentThread().interrupt(); // Restore interrupt status 508 failures.add(file); 509 } catch (ExecutionException e) { 510 LOG.error("Archiving failed for file: {}", file, e); 511 failures.add(file); 512 } 513 } 514 } finally { 515 hfilesArchiveExecutor.shutdown(); 516 } 517 } 518 519 /** 520 * Attempt to archive the passed in file to the archive directory. 521 * <p> 522 * If the same file already exists in the archive, it is moved to a timestamped directory under 523 * the archive directory and the new file is put in its place. 524 * @param archiveDir {@link Path} to the directory that stores the archives of the hfiles 525 * @param currentFile {@link Path} to the original HFile that will be archived 526 * @param archiveStartTime time the archiving started, to resolve naming conflicts 527 * @return <tt>true</tt> if the file is successfully archived. <tt>false</tt> if there was a 528 * problem, but the operation still completed. 529 * @throws IOException on failure to complete {@link FileSystem} operations. 530 */ 531 private static boolean resolveAndArchiveFile(Path archiveDir, File currentFile, 532 String archiveStartTime) throws IOException { 533 // build path as it should be in the archive 534 String filename = currentFile.getName(); 535 Path archiveFile = new Path(archiveDir, filename); 536 FileSystem fs = currentFile.getFileSystem(); 537 538 // An existing destination file in the archive is unexpected, but we handle it here. 539 if (fs.exists(archiveFile)) { 540 if (!fs.exists(currentFile.getPath())) { 541 // If the file already exists in the archive, and there is no current file to archive, then 542 // assume that the file in archive is correct. This is an unexpected situation, suggesting a 543 // race condition or split brain. 544 // In HBASE-26718 this was found when compaction incorrectly happened during warmupRegion. 545 LOG.warn("{} exists in archive. Attempted to archive nonexistent file {}.", archiveFile, 546 currentFile); 547 // We return success to match existing behavior in this method, where FileNotFoundException 548 // in moveAndClose is ignored. 549 return true; 550 } 551 // There is a conflict between the current file and the already existing archived file. 552 // Move the archived file to a timestamped backup. This is a really, really unlikely 553 // situation, where we get the same name for the existing file, but is included just for that 554 // 1 in trillion chance. We are potentially incurring data loss in the archive directory if 555 // the files are not identical. The timestamped backup will be cleaned by HFileCleaner as it 556 // has no references. 557 FileStatus curStatus = fs.getFileStatus(currentFile.getPath()); 558 FileStatus archiveStatus = fs.getFileStatus(archiveFile); 559 long curLen = curStatus.getLen(); 560 long archiveLen = archiveStatus.getLen(); 561 long curMtime = curStatus.getModificationTime(); 562 long archiveMtime = archiveStatus.getModificationTime(); 563 if (curLen != archiveLen) { 564 LOG.error( 565 "{} already exists in archive with different size than current {}." 566 + " archiveLen: {} currentLen: {} archiveMtime: {} currentMtime: {}", 567 archiveFile, currentFile, archiveLen, curLen, archiveMtime, curMtime); 568 throw new IOException( 569 archiveFile + " already exists in archive with different size" + " than " + currentFile); 570 } 571 572 LOG.error( 573 "{} already exists in archive, moving to timestamped backup and overwriting" 574 + " current {}. archiveLen: {} currentLen: {} archiveMtime: {} currentMtime: {}", 575 archiveFile, currentFile, archiveLen, curLen, archiveMtime, curMtime); 576 577 // move the archive file to the stamped backup 578 Path backedupArchiveFile = new Path(archiveDir, filename + SEPARATOR + archiveStartTime); 579 if (!fs.rename(archiveFile, backedupArchiveFile)) { 580 LOG.error("Could not rename archive file to backup: " + backedupArchiveFile 581 + ", deleting existing file in favor of newer."); 582 // try to delete the existing file, if we can't rename it 583 if (!fs.delete(archiveFile, false)) { 584 throw new IOException("Couldn't delete existing archive file (" + archiveFile 585 + ") or rename it to the backup file (" + backedupArchiveFile 586 + ") to make room for similarly named file."); 587 } 588 } else { 589 LOG.info("Backed up archive file from {} to {}.", archiveFile, backedupArchiveFile); 590 } 591 } 592 593 LOG.trace("No existing file in archive for {}, free to archive original file.", archiveFile); 594 595 // at this point, we should have a free spot for the archive file 596 boolean success = false; 597 for (int i = 0; !success && i < DEFAULT_RETRIES_NUMBER; ++i) { 598 if (i > 0) { 599 // Ensure that the archive directory exists. 600 // The previous "move to archive" operation has failed probably because 601 // the cleaner has removed our archive directory (HBASE-7643). 602 // (we're in a retry loop, so don't worry too much about the exception) 603 try { 604 if (!fs.exists(archiveDir)) { 605 if (fs.mkdirs(archiveDir)) { 606 LOG.debug("Created archive directory {}", archiveDir); 607 } 608 } 609 } catch (IOException e) { 610 LOG.warn("Failed to create directory {}", archiveDir, e); 611 } 612 } 613 614 try { 615 success = currentFile.moveAndClose(archiveFile); 616 } catch (FileNotFoundException fnfe) { 617 LOG.warn("Failed to archive " + currentFile 618 + " because it does not exist! Skipping and continuing on.", fnfe); 619 success = true; 620 } catch (IOException e) { 621 success = false; 622 // When HFiles are placed on a filesystem other than HDFS a rename operation can be a 623 // non-atomic file copy operation. It can take a long time to copy a large hfile and if 624 // interrupted there may be a partially copied file present at the destination. We must 625 // remove the partially copied file, if any, or otherwise the archive operation will fail 626 // indefinitely from this point. 627 LOG.warn("Failed to archive " + currentFile + " on try #" + i, e); 628 try { 629 fs.delete(archiveFile, false); 630 } catch (FileNotFoundException fnfe) { 631 // This case is fine. 632 } catch (IOException ee) { 633 // Complain about other IO exceptions 634 LOG.warn("Failed to clean up from failure to archive " + currentFile + " on try #" + i, 635 ee); 636 } 637 } 638 } 639 640 if (!success) { 641 LOG.error("Failed to archive " + currentFile); 642 return false; 643 } 644 645 LOG.debug("Archived from {} to {}", currentFile, archiveFile); 646 return true; 647 } 648 649 /** 650 * Without regard for backup, delete a region. Should be used with caution. 651 * @param regionDir {@link Path} to the region to be deleted. 652 * @param fs FileSystem from which to delete the region 653 * @return <tt>true</tt> on successful deletion, <tt>false</tt> otherwise 654 * @throws IOException on filesystem operation failure 655 */ 656 private static boolean deleteRegionWithoutArchiving(FileSystem fs, Path regionDir) 657 throws IOException { 658 if (fs.delete(regionDir, true)) { 659 LOG.debug("Deleted {}", regionDir); 660 return true; 661 } 662 LOG.debug("Failed to delete directory {}", regionDir); 663 return false; 664 } 665 666 /** 667 * Just do a simple delete of the given store files 668 * <p> 669 * A best effort is made to delete each of the files, rather than bailing on the first failure. 670 * <p> 671 * @param compactedFiles store files to delete from the file system. 672 * @throws IOException if a file cannot be deleted. All files will be attempted to deleted before 673 * throwing the exception, rather than failing at the first file. 674 */ 675 private static void deleteStoreFilesWithoutArchiving(Collection<HStoreFile> compactedFiles) 676 throws IOException { 677 LOG.debug("Deleting files without archiving."); 678 List<IOException> errors = new ArrayList<>(0); 679 for (HStoreFile hsf : compactedFiles) { 680 try { 681 hsf.deleteStoreFile(); 682 } catch (IOException e) { 683 LOG.error("Failed to delete {}", hsf.getPath()); 684 errors.add(e); 685 } 686 } 687 if (errors.size() > 0) { 688 throw MultipleIOException.createIOException(errors); 689 } 690 } 691 692 /** 693 * Adapt a type to match the {@link File} interface, which is used internally for handling 694 * archival/removal of files 695 * @param <T> type to adapt to the {@link File} interface 696 */ 697 private static abstract class FileConverter<T> implements Function<T, File> { 698 protected final FileSystem fs; 699 700 public FileConverter(FileSystem fs) { 701 this.fs = fs; 702 } 703 } 704 705 /** 706 * Convert a FileStatus to something we can manage in the archiving 707 */ 708 private static class FileStatusConverter extends FileConverter<FileStatus> { 709 public FileStatusConverter(FileSystem fs) { 710 super(fs); 711 } 712 713 @Override 714 public File apply(FileStatus input) { 715 return new FileablePath(fs, input.getPath()); 716 } 717 } 718 719 /** 720 * Convert the {@link HStoreFile} into something we can manage in the archive methods 721 */ 722 private static class StoreToFile extends FileConverter<HStoreFile> { 723 public StoreToFile(FileSystem fs) { 724 super(fs); 725 } 726 727 @Override 728 public File apply(HStoreFile input) { 729 return new FileableStoreFile(fs, input); 730 } 731 } 732 733 /** 734 * Wrapper to handle file operations uniformly 735 */ 736 private static abstract class File { 737 protected final FileSystem fs; 738 739 public File(FileSystem fs) { 740 this.fs = fs; 741 } 742 743 /** 744 * Delete the file 745 * @throws IOException on failure 746 */ 747 abstract void delete() throws IOException; 748 749 /** 750 * Check to see if this is a file or a directory 751 * @return <tt>true</tt> if it is a file, <tt>false</tt> otherwise 752 * @throws IOException on {@link FileSystem} connection error 753 */ 754 abstract boolean isFile() throws IOException; 755 756 /** 757 * @return if this is a directory, returns all the children in the directory, otherwise returns 758 * an empty list 759 */ 760 abstract Collection<File> getChildren() throws IOException; 761 762 /** 763 * close any outside readers of the file 764 */ 765 abstract void close() throws IOException; 766 767 /** Returns the name of the file (not the full fs path, just the individual file name) */ 768 abstract String getName(); 769 770 /** Returns the path to this file */ 771 abstract Path getPath(); 772 773 /** 774 * Move the file to the given destination 775 * @return <tt>true</tt> on success 776 */ 777 public boolean moveAndClose(Path dest) throws IOException { 778 this.close(); 779 Path p = this.getPath(); 780 return CommonFSUtils.renameAndSetModifyTime(fs, p, dest); 781 } 782 783 /** Returns the {@link FileSystem} on which this file resides */ 784 public FileSystem getFileSystem() { 785 return this.fs; 786 } 787 788 @Override 789 public String toString() { 790 return this.getClass().getSimpleName() + ", " + getPath().toString(); 791 } 792 } 793 794 /** 795 * A {@link File} that wraps a simple {@link Path} on a {@link FileSystem}. 796 */ 797 private static class FileablePath extends File { 798 private final Path file; 799 private final FileStatusConverter getAsFile; 800 801 public FileablePath(FileSystem fs, Path file) { 802 super(fs); 803 this.file = file; 804 this.getAsFile = new FileStatusConverter(fs); 805 } 806 807 @Override 808 public void delete() throws IOException { 809 if (!fs.delete(file, true)) throw new IOException("Failed to delete:" + this.file); 810 } 811 812 @Override 813 public String getName() { 814 return file.getName(); 815 } 816 817 @Override 818 public Collection<File> getChildren() throws IOException { 819 if (fs.isFile(file)) { 820 return Collections.emptyList(); 821 } 822 return Stream.of(fs.listStatus(file)).map(getAsFile).collect(Collectors.toList()); 823 } 824 825 @Override 826 public boolean isFile() throws IOException { 827 return fs.isFile(file); 828 } 829 830 @Override 831 public void close() throws IOException { 832 // NOOP - files are implicitly closed on removal 833 } 834 835 @Override 836 Path getPath() { 837 return file; 838 } 839 } 840 841 /** 842 * {@link File} adapter for a {@link HStoreFile} living on a {@link FileSystem} . 843 */ 844 private static class FileableStoreFile extends File { 845 HStoreFile file; 846 847 public FileableStoreFile(FileSystem fs, HStoreFile store) { 848 super(fs); 849 this.file = store; 850 } 851 852 @Override 853 public void delete() throws IOException { 854 file.deleteStoreFile(); 855 } 856 857 @Override 858 public String getName() { 859 return file.getPath().getName(); 860 } 861 862 @Override 863 public boolean isFile() { 864 return true; 865 } 866 867 @Override 868 public Collection<File> getChildren() throws IOException { 869 // storefiles don't have children 870 return Collections.emptyList(); 871 } 872 873 @Override 874 public void close() throws IOException { 875 file.closeStoreFile(true); 876 } 877 878 @Override 879 Path getPath() { 880 return file.getPath(); 881 } 882 } 883}