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 */ 018 019package org.apache.hadoop.fs.azure; 020 021import java.io.DataInputStream; 022import java.io.FileNotFoundException; 023import java.io.IOException; 024import java.io.InputStream; 025import java.io.OutputStream; 026import java.net.URI; 027import java.net.URISyntaxException; 028import java.text.SimpleDateFormat; 029import java.util.ArrayList; 030import java.util.Date; 031import java.util.EnumSet; 032import java.util.Iterator; 033import java.util.Set; 034import java.util.TimeZone; 035import java.util.TreeSet; 036import java.util.UUID; 037import java.util.concurrent.atomic.AtomicInteger; 038import java.util.regex.Matcher; 039import java.util.regex.Pattern; 040 041import org.apache.commons.lang.StringUtils; 042import org.apache.commons.lang.exception.ExceptionUtils; 043import org.apache.commons.logging.Log; 044import org.apache.commons.logging.LogFactory; 045import org.apache.hadoop.classification.InterfaceAudience; 046import org.apache.hadoop.classification.InterfaceStability; 047import org.apache.hadoop.conf.Configuration; 048import org.apache.hadoop.fs.BlockLocation; 049import org.apache.hadoop.fs.BufferedFSInputStream; 050import org.apache.hadoop.fs.CreateFlag; 051import org.apache.hadoop.fs.FSDataInputStream; 052import org.apache.hadoop.fs.FSDataOutputStream; 053import org.apache.hadoop.fs.FSInputStream; 054import org.apache.hadoop.fs.FileStatus; 055import org.apache.hadoop.fs.FileSystem; 056import org.apache.hadoop.fs.Path; 057import org.apache.hadoop.fs.azure.metrics.AzureFileSystemInstrumentation; 058import org.apache.hadoop.fs.azure.metrics.AzureFileSystemMetricsSystem; 059import org.apache.hadoop.fs.permission.FsPermission; 060import org.apache.hadoop.fs.permission.PermissionStatus; 061import org.apache.hadoop.fs.azure.AzureException; 062import org.apache.hadoop.fs.azure.StorageInterface.CloudBlobWrapper; 063import org.apache.hadoop.io.IOUtils; 064import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem; 065import org.apache.hadoop.security.UserGroupInformation; 066import org.apache.hadoop.util.Progressable; 067 068 069import org.codehaus.jackson.JsonNode; 070import org.codehaus.jackson.JsonParseException; 071import org.codehaus.jackson.JsonParser; 072import org.codehaus.jackson.map.JsonMappingException; 073import org.codehaus.jackson.map.ObjectMapper; 074 075import com.google.common.annotations.VisibleForTesting; 076import com.microsoft.windowsazure.storage.AccessCondition; 077import com.microsoft.windowsazure.storage.OperationContext; 078import com.microsoft.windowsazure.storage.StorageException; 079import com.microsoft.windowsazure.storage.blob.CloudBlob; 080import com.microsoft.windowsazure.storage.core.*; 081 082/** 083 * <p> 084 * A {@link FileSystem} for reading and writing files stored on <a 085 * href="http://store.azure.com/">Windows Azure</a>. This implementation is 086 * blob-based and stores files on Azure in their native form so they can be read 087 * by other Azure tools. 088 * </p> 089 */ 090@InterfaceAudience.Public 091@InterfaceStability.Stable 092public class NativeAzureFileSystem extends FileSystem { 093 private static final int USER_WX_PERMISION = 0300; 094 095 /** 096 * A description of a folder rename operation, including the source and 097 * destination keys, and descriptions of the files in the source folder. 098 */ 099 public static class FolderRenamePending { 100 private SelfRenewingLease folderLease; 101 private String srcKey; 102 private String dstKey; 103 private FileMetadata[] fileMetadata = null; // descriptions of source files 104 private ArrayList<String> fileStrings = null; 105 private NativeAzureFileSystem fs; 106 private static final int MAX_RENAME_PENDING_FILE_SIZE = 10000000; 107 private static final int FORMATTING_BUFFER = 10000; 108 private boolean committed; 109 public static final String SUFFIX = "-RenamePending.json"; 110 111 // Prepare in-memory information needed to do or redo a folder rename. 112 public FolderRenamePending(String srcKey, String dstKey, SelfRenewingLease lease, 113 NativeAzureFileSystem fs) throws IOException { 114 this.srcKey = srcKey; 115 this.dstKey = dstKey; 116 this.folderLease = lease; 117 this.fs = fs; 118 ArrayList<FileMetadata> fileMetadataList = new ArrayList<FileMetadata>(); 119 120 // List all the files in the folder. 121 String priorLastKey = null; 122 do { 123 PartialListing listing = fs.getStoreInterface().listAll(srcKey, AZURE_LIST_ALL, 124 AZURE_UNBOUNDED_DEPTH, priorLastKey); 125 for(FileMetadata file : listing.getFiles()) { 126 fileMetadataList.add(file); 127 } 128 priorLastKey = listing.getPriorLastKey(); 129 } while (priorLastKey != null); 130 fileMetadata = fileMetadataList.toArray(new FileMetadata[fileMetadataList.size()]); 131 this.committed = true; 132 } 133 134 // Prepare in-memory information needed to do or redo folder rename from 135 // a -RenamePending.json file read from storage. This constructor is to use during 136 // redo processing. 137 public FolderRenamePending(Path redoFile, NativeAzureFileSystem fs) 138 throws IllegalArgumentException, IOException { 139 140 this.fs = fs; 141 142 // open redo file 143 Path f = redoFile; 144 FSDataInputStream input = fs.open(f); 145 byte[] bytes = new byte[MAX_RENAME_PENDING_FILE_SIZE]; 146 int l = input.read(bytes); 147 if (l < 0) { 148 throw new IOException( 149 "Error reading pending rename file contents -- no data available"); 150 } 151 if (l == MAX_RENAME_PENDING_FILE_SIZE) { 152 throw new IOException( 153 "Error reading pending rename file contents -- " 154 + "maximum file size exceeded"); 155 } 156 String contents = new String(bytes, 0, l); 157 158 // parse the JSON 159 ObjectMapper objMapper = new ObjectMapper(); 160 objMapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES, true); 161 JsonNode json = null; 162 try { 163 json = objMapper.readValue(contents, JsonNode.class); 164 this.committed = true; 165 } catch (JsonMappingException e) { 166 167 // The -RedoPending.json file is corrupted, so we assume it was 168 // not completely written 169 // and the redo operation did not commit. 170 this.committed = false; 171 } catch (JsonParseException e) { 172 this.committed = false; 173 } catch (IOException e) { 174 this.committed = false; 175 } 176 177 if (!this.committed) { 178 LOG.error("Deleting corruped rename pending file " 179 + redoFile + "\n" + contents); 180 181 // delete the -RenamePending.json file 182 fs.delete(redoFile, false); 183 return; 184 } 185 186 // initialize this object's fields 187 ArrayList<String> fileStrList = new ArrayList<String>(); 188 JsonNode oldFolderName = json.get("OldFolderName"); 189 JsonNode newFolderName = json.get("NewFolderName"); 190 if (oldFolderName == null || newFolderName == null) { 191 this.committed = false; 192 } else { 193 this.srcKey = oldFolderName.getTextValue(); 194 this.dstKey = newFolderName.getTextValue(); 195 if (this.srcKey == null || this.dstKey == null) { 196 this.committed = false; 197 } else { 198 JsonNode fileList = json.get("FileList"); 199 if (fileList == null) { 200 this.committed = false; 201 } else { 202 for (int i = 0; i < fileList.size(); i++) { 203 fileStrList.add(fileList.get(i).getTextValue()); 204 } 205 } 206 } 207 } 208 this.fileStrings = fileStrList; 209 } 210 211 public FileMetadata[] getFiles() { 212 return fileMetadata; 213 } 214 215 public SelfRenewingLease getFolderLease() { 216 return folderLease; 217 } 218 219 /** 220 * Write to disk the information needed to redo folder rename, in JSON format. 221 * The file name will be wasb://<sourceFolderPrefix>/folderName-RenamePending.json 222 * The file format will be: 223 * { 224 * FormatVersion: "1.0", 225 * OperationTime: "<YYYY-MM-DD HH:MM:SS.MMM>", 226 * OldFolderName: "<key>", 227 * NewFolderName: "<key>", 228 * FileList: [ <string> , <string> , ... ] 229 * } 230 * 231 * Here's a sample: 232 * { 233 * FormatVersion: "1.0", 234 * OperationUTCTime: "2014-07-01 23:50:35.572", 235 * OldFolderName: "user/ehans/folderToRename", 236 * NewFolderName: "user/ehans/renamedFolder", 237 * FileList: [ 238 * "innerFile", 239 * "innerFile2" 240 * ] 241 * } 242 * @throws IOException 243 */ 244 public void writeFile(FileSystem fs) throws IOException { 245 Path path = getRenamePendingFilePath(); 246 if (LOG.isDebugEnabled()){ 247 LOG.debug("Preparing to write atomic rename state to " + path.toString()); 248 } 249 OutputStream output = null; 250 251 String contents = makeRenamePendingFileContents(); 252 253 // Write file. 254 try { 255 output = fs.create(path); 256 output.write(contents.getBytes()); 257 } catch (IOException e) { 258 throw new IOException("Unable to write RenamePending file for folder rename from " 259 + srcKey + " to " + dstKey, e); 260 } finally { 261 IOUtils.cleanup(LOG, output); 262 } 263 } 264 265 /** 266 * Return the contents of the JSON file to represent the operations 267 * to be performed for a folder rename. 268 */ 269 public String makeRenamePendingFileContents() { 270 SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS"); 271 sdf.setTimeZone(TimeZone.getTimeZone("UTC")); 272 String time = sdf.format(new Date()); 273 274 // Make file list string 275 StringBuilder builder = new StringBuilder(); 276 builder.append("[\n"); 277 for (int i = 0; i != fileMetadata.length; i++) { 278 if (i > 0) { 279 builder.append(",\n"); 280 } 281 builder.append(" "); 282 String noPrefix = StringUtils.removeStart(fileMetadata[i].getKey(), srcKey + "/"); 283 284 // Quote string file names, escaping any possible " characters or other 285 // necessary characters in the name. 286 builder.append(quote(noPrefix)); 287 if (builder.length() >= 288 MAX_RENAME_PENDING_FILE_SIZE - FORMATTING_BUFFER) { 289 290 // Give up now to avoid using too much memory. 291 LOG.error("Internal error: Exceeded maximum rename pending file size of " 292 + MAX_RENAME_PENDING_FILE_SIZE + " bytes."); 293 294 // return some bad JSON with an error message to make it human readable 295 return "exceeded maximum rename pending file size"; 296 } 297 } 298 builder.append("\n ]"); 299 String fileList = builder.toString(); 300 301 // Make file contents as a string. Again, quote file names, escaping 302 // characters as appropriate. 303 String contents = "{\n" 304 + " FormatVersion: \"1.0\",\n" 305 + " OperationUTCTime: \"" + time + "\",\n" 306 + " OldFolderName: " + quote(srcKey) + ",\n" 307 + " NewFolderName: " + quote(dstKey) + ",\n" 308 + " FileList: " + fileList + "\n" 309 + "}\n"; 310 311 return contents; 312 } 313 314 /** 315 * This is an exact copy of org.codehaus.jettison.json.JSONObject.quote 316 * method. 317 * 318 * Produce a string in double quotes with backslash sequences in all the 319 * right places. A backslash will be inserted within </, allowing JSON 320 * text to be delivered in HTML. In JSON text, a string cannot contain a 321 * control character or an unescaped quote or backslash. 322 * @param string A String 323 * @return A String correctly formatted for insertion in a JSON text. 324 */ 325 private String quote(String string) { 326 if (string == null || string.length() == 0) { 327 return "\"\""; 328 } 329 330 char c = 0; 331 int i; 332 int len = string.length(); 333 StringBuilder sb = new StringBuilder(len + 4); 334 String t; 335 336 sb.append('"'); 337 for (i = 0; i < len; i += 1) { 338 c = string.charAt(i); 339 switch (c) { 340 case '\\': 341 case '"': 342 sb.append('\\'); 343 sb.append(c); 344 break; 345 case '/': 346 sb.append('\\'); 347 sb.append(c); 348 break; 349 case '\b': 350 sb.append("\\b"); 351 break; 352 case '\t': 353 sb.append("\\t"); 354 break; 355 case '\n': 356 sb.append("\\n"); 357 break; 358 case '\f': 359 sb.append("\\f"); 360 break; 361 case '\r': 362 sb.append("\\r"); 363 break; 364 default: 365 if (c < ' ') { 366 t = "000" + Integer.toHexString(c); 367 sb.append("\\u" + t.substring(t.length() - 4)); 368 } else { 369 sb.append(c); 370 } 371 } 372 } 373 sb.append('"'); 374 return sb.toString(); 375 } 376 377 public String getSrcKey() { 378 return srcKey; 379 } 380 381 public String getDstKey() { 382 return dstKey; 383 } 384 385 public FileMetadata getSourceMetadata() throws IOException { 386 return fs.getStoreInterface().retrieveMetadata(srcKey); 387 } 388 389 /** 390 * Execute a folder rename. This is the execution path followed 391 * when everything is working normally. See redo() for the alternate 392 * execution path for the case where we're recovering from a folder rename 393 * failure. 394 * @throws IOException 395 */ 396 public void execute() throws IOException { 397 398 for (FileMetadata file : this.getFiles()) { 399 400 // Rename all materialized entries under the folder to point to the 401 // final destination. 402 if (file.getBlobMaterialization() == BlobMaterialization.Explicit) { 403 String srcName = file.getKey(); 404 String suffix = srcName.substring((this.getSrcKey()).length()); 405 String dstName = this.getDstKey() + suffix; 406 407 // Rename gets exclusive access (via a lease) for files 408 // designated for atomic rename. 409 // The main use case is for HBase write-ahead log (WAL) and data 410 // folder processing correctness. See the rename code for details. 411 boolean acquireLease = fs.getStoreInterface().isAtomicRenameKey(srcName); 412 fs.getStoreInterface().rename(srcName, dstName, acquireLease, null); 413 } 414 } 415 416 // Rename the source folder 0-byte root file itself. 417 FileMetadata srcMetadata2 = this.getSourceMetadata(); 418 if (srcMetadata2.getBlobMaterialization() == 419 BlobMaterialization.Explicit) { 420 421 // It already has a lease on it from the "prepare" phase so there's no 422 // need to get one now. Pass in existing lease to allow file delete. 423 fs.getStoreInterface().rename(this.getSrcKey(), this.getDstKey(), 424 false, folderLease); 425 } 426 427 // Update the last-modified time of the parent folders of both source and 428 // destination. 429 fs.updateParentFolderLastModifiedTime(srcKey); 430 fs.updateParentFolderLastModifiedTime(dstKey); 431 } 432 433 /** Clean up after execution of rename. 434 * @throws IOException */ 435 public void cleanup() throws IOException { 436 437 if (fs.getStoreInterface().isAtomicRenameKey(srcKey)) { 438 439 // Remove RenamePending file 440 fs.delete(getRenamePendingFilePath(), false); 441 442 // Freeing source folder lease is not necessary since the source 443 // folder file was deleted. 444 } 445 } 446 447 private Path getRenamePendingFilePath() { 448 String fileName = srcKey + SUFFIX; 449 Path fileNamePath = keyToPath(fileName); 450 Path path = fs.makeAbsolute(fileNamePath); 451 return path; 452 } 453 454 /** 455 * Recover from a folder rename failure by redoing the intended work, 456 * as recorded in the -RenamePending.json file. 457 * 458 * @throws IOException 459 */ 460 public void redo() throws IOException { 461 462 if (!committed) { 463 464 // Nothing to do. The -RedoPending.json file should have already been 465 // deleted. 466 return; 467 } 468 469 // Try to get a lease on source folder to block concurrent access to it. 470 // It may fail if the folder is already gone. We don't check if the 471 // source exists explicitly because that could recursively trigger redo 472 // and give an infinite recursion. 473 SelfRenewingLease lease = null; 474 boolean sourceFolderGone = false; 475 try { 476 lease = fs.leaseSourceFolder(srcKey); 477 } catch (AzureException e) { 478 479 // If the source folder was not found then somebody probably 480 // raced with us and finished the rename first, or the 481 // first rename failed right before deleting the rename pending 482 // file. 483 String errorCode = ""; 484 try { 485 StorageException se = (StorageException) e.getCause(); 486 errorCode = se.getErrorCode(); 487 } catch (Exception e2) { 488 ; // do nothing -- could not get errorCode 489 } 490 if (errorCode.equals("BlobNotFound")) { 491 sourceFolderGone = true; 492 } else { 493 throw new IOException( 494 "Unexpected error when trying to lease source folder name during " 495 + "folder rename redo", 496 e); 497 } 498 } 499 500 if (!sourceFolderGone) { 501 // Make sure the target folder exists. 502 Path dst = fullPath(dstKey); 503 if (!fs.exists(dst)) { 504 fs.mkdirs(dst); 505 } 506 507 // For each file inside the folder to be renamed, 508 // make sure it has been renamed. 509 for(String fileName : fileStrings) { 510 finishSingleFileRename(fileName); 511 } 512 513 // Remove the source folder. Don't check explicitly if it exists, 514 // to avoid triggering redo recursively. 515 try { 516 fs.getStoreInterface().delete(srcKey, lease); 517 } catch (Exception e) { 518 LOG.info("Unable to delete source folder during folder rename redo. " 519 + "If the source folder is already gone, this is not an error " 520 + "condition. Continuing with redo.", e); 521 } 522 523 // Update the last-modified time of the parent folders of both source 524 // and destination. 525 fs.updateParentFolderLastModifiedTime(srcKey); 526 fs.updateParentFolderLastModifiedTime(dstKey); 527 } 528 529 // Remove the -RenamePending.json file. 530 fs.delete(getRenamePendingFilePath(), false); 531 } 532 533 // See if the source file is still there, and if it is, rename it. 534 private void finishSingleFileRename(String fileName) 535 throws IOException { 536 Path srcFile = fullPath(srcKey, fileName); 537 Path dstFile = fullPath(dstKey, fileName); 538 boolean srcExists = fs.exists(srcFile); 539 boolean dstExists = fs.exists(dstFile); 540 if (srcExists && !dstExists) { 541 542 // Rename gets exclusive access (via a lease) for HBase write-ahead log 543 // (WAL) file processing correctness. See the rename code for details. 544 String srcName = fs.pathToKey(srcFile); 545 String dstName = fs.pathToKey(dstFile); 546 fs.getStoreInterface().rename(srcName, dstName, true, null); 547 } else if (srcExists && dstExists) { 548 549 // Get a lease on source to block write access. 550 String srcName = fs.pathToKey(srcFile); 551 SelfRenewingLease lease = fs.acquireLease(srcFile); 552 553 // Delete the file. This will free the lease too. 554 fs.getStoreInterface().delete(srcName, lease); 555 } else if (!srcExists && dstExists) { 556 557 // The rename already finished, so do nothing. 558 ; 559 } else { 560 throw new IOException( 561 "Attempting to complete rename of file " + srcKey + "/" + fileName 562 + " during folder rename redo, and file was not found in source " 563 + "or destination."); 564 } 565 } 566 567 // Return an absolute path for the specific fileName within the folder 568 // specified by folderKey. 569 private Path fullPath(String folderKey, String fileName) { 570 return new Path(new Path(fs.getUri()), "/" + folderKey + "/" + fileName); 571 } 572 573 private Path fullPath(String fileKey) { 574 return new Path(new Path(fs.getUri()), "/" + fileKey); 575 } 576 } 577 578 private static final String TRAILING_PERIOD_PLACEHOLDER = "[[.]]"; 579 private static final Pattern TRAILING_PERIOD_PLACEHOLDER_PATTERN = 580 Pattern.compile("\\[\\[\\.\\]\\](?=$|/)"); 581 private static final Pattern TRAILING_PERIOD_PATTERN = Pattern.compile("\\.(?=$|/)"); 582 583 @Override 584 public String getScheme() { 585 return "wasb"; 586 } 587 588 589 /** 590 * <p> 591 * A {@link FileSystem} for reading and writing files stored on <a 592 * href="http://store.azure.com/">Windows Azure</a>. This implementation is 593 * blob-based and stores files on Azure in their native form so they can be read 594 * by other Azure tools. This implementation uses HTTPS for secure network communication. 595 * </p> 596 */ 597 public static class Secure extends NativeAzureFileSystem { 598 @Override 599 public String getScheme() { 600 return "wasbs"; 601 } 602 } 603 604 public static final Log LOG = LogFactory.getLog(NativeAzureFileSystem.class); 605 606 static final String AZURE_BLOCK_SIZE_PROPERTY_NAME = "fs.azure.block.size"; 607 /** 608 * The time span in seconds before which we consider a temp blob to be 609 * dangling (not being actively uploaded to) and up for reclamation. 610 * 611 * So e.g. if this is 60, then any temporary blobs more than a minute old 612 * would be considered dangling. 613 */ 614 static final String AZURE_TEMP_EXPIRY_PROPERTY_NAME = "fs.azure.fsck.temp.expiry.seconds"; 615 private static final int AZURE_TEMP_EXPIRY_DEFAULT = 3600; 616 static final String PATH_DELIMITER = Path.SEPARATOR; 617 static final String AZURE_TEMP_FOLDER = "_$azuretmpfolder$"; 618 619 private static final int AZURE_LIST_ALL = -1; 620 private static final int AZURE_UNBOUNDED_DEPTH = -1; 621 622 private static final long MAX_AZURE_BLOCK_SIZE = 512 * 1024 * 1024L; 623 624 /** 625 * The configuration property that determines which group owns files created 626 * in WASB. 627 */ 628 private static final String AZURE_DEFAULT_GROUP_PROPERTY_NAME = "fs.azure.permissions.supergroup"; 629 /** 630 * The default value for fs.azure.permissions.supergroup. Chosen as the same 631 * default as DFS. 632 */ 633 static final String AZURE_DEFAULT_GROUP_DEFAULT = "supergroup"; 634 635 static final String AZURE_BLOCK_LOCATION_HOST_PROPERTY_NAME = 636 "fs.azure.block.location.impersonatedhost"; 637 private static final String AZURE_BLOCK_LOCATION_HOST_DEFAULT = 638 "localhost"; 639 static final String AZURE_RINGBUFFER_CAPACITY_PROPERTY_NAME = 640 "fs.azure.ring.buffer.capacity"; 641 static final String AZURE_OUTPUT_STREAM_BUFFER_SIZE_PROPERTY_NAME = 642 "fs.azure.output.stream.buffer.size"; 643 644 private class NativeAzureFsInputStream extends FSInputStream { 645 private InputStream in; 646 private final String key; 647 private long pos = 0; 648 private boolean closed = false; 649 private boolean isPageBlob; 650 651 // File length, valid only for streams over block blobs. 652 private long fileLength; 653 654 public NativeAzureFsInputStream(DataInputStream in, String key, long fileLength) { 655 this.in = in; 656 this.key = key; 657 this.isPageBlob = store.isPageBlobKey(key); 658 this.fileLength = fileLength; 659 } 660 661 /** 662 * Return the size of the remaining available bytes 663 * if the size is less than or equal to {@link Integer#MAX_VALUE}, 664 * otherwise, return {@link Integer#MAX_VALUE}. 665 * 666 * This is to match the behavior of DFSInputStream.available(), 667 * which some clients may rely on (HBase write-ahead log reading in 668 * particular). 669 */ 670 @Override 671 public synchronized int available() throws IOException { 672 if (isPageBlob) { 673 return in.available(); 674 } else { 675 if (closed) { 676 throw new IOException("Stream closed"); 677 } 678 final long remaining = this.fileLength - pos; 679 return remaining <= Integer.MAX_VALUE ? 680 (int) remaining : Integer.MAX_VALUE; 681 } 682 } 683 684 /* 685 * Reads the next byte of data from the input stream. The value byte is 686 * returned as an integer in the range 0 to 255. If no byte is available 687 * because the end of the stream has been reached, the value -1 is returned. 688 * This method blocks until input data is available, the end of the stream 689 * is detected, or an exception is thrown. 690 * 691 * @returns int An integer corresponding to the byte read. 692 */ 693 @Override 694 public synchronized int read() throws IOException { 695 int result = 0; 696 result = in.read(); 697 if (result != -1) { 698 pos++; 699 if (statistics != null) { 700 statistics.incrementBytesRead(1); 701 } 702 } 703 704 // Return to the caller with the result. 705 // 706 return result; 707 } 708 709 /* 710 * Reads up to len bytes of data from the input stream into an array of 711 * bytes. An attempt is made to read as many as len bytes, but a smaller 712 * number may be read. The number of bytes actually read is returned as an 713 * integer. This method blocks until input data is available, end of file is 714 * detected, or an exception is thrown. If len is zero, then no bytes are 715 * read and 0 is returned; otherwise, there is an attempt to read at least 716 * one byte. If no byte is available because the stream is at end of file, 717 * the value -1 is returned; otherwise, at least one byte is read and stored 718 * into b. 719 * 720 * @param b -- the buffer into which data is read 721 * 722 * @param off -- the start offset in the array b at which data is written 723 * 724 * @param len -- the maximum number of bytes read 725 * 726 * @ returns int The total number of byes read into the buffer, or -1 if 727 * there is no more data because the end of stream is reached. 728 */ 729 @Override 730 public synchronized int read(byte[] b, int off, int len) throws IOException { 731 int result = 0; 732 result = in.read(b, off, len); 733 if (result > 0) { 734 pos += result; 735 } 736 737 if (null != statistics) { 738 statistics.incrementBytesRead(result); 739 } 740 741 // Return to the caller with the result. 742 return result; 743 } 744 745 @Override 746 public void close() throws IOException { 747 in.close(); 748 closed = true; 749 } 750 751 @Override 752 public synchronized void seek(long pos) throws IOException { 753 in.close(); 754 in = store.retrieve(key); 755 this.pos = in.skip(pos); 756 if (LOG.isDebugEnabled()) { 757 LOG.debug(String.format("Seek to position %d. Bytes skipped %d", pos, 758 this.pos)); 759 } 760 } 761 762 @Override 763 public synchronized long getPos() throws IOException { 764 return pos; 765 } 766 767 @Override 768 public boolean seekToNewSource(long targetPos) throws IOException { 769 return false; 770 } 771 } 772 773 private class NativeAzureFsOutputStream extends OutputStream { 774 // We should not override flush() to actually close current block and flush 775 // to DFS, this will break applications that assume flush() is a no-op. 776 // Applications are advised to use Syncable.hflush() for that purpose. 777 // NativeAzureFsOutputStream needs to implement Syncable if needed. 778 private String key; 779 private String keyEncoded; 780 private OutputStream out; 781 782 public NativeAzureFsOutputStream(OutputStream out, String aKey, 783 String anEncodedKey) throws IOException { 784 // Check input arguments. The output stream should be non-null and the 785 // keys 786 // should be valid strings. 787 if (null == out) { 788 throw new IllegalArgumentException( 789 "Illegal argument: the output stream is null."); 790 } 791 792 if (null == aKey || 0 == aKey.length()) { 793 throw new IllegalArgumentException( 794 "Illegal argument the key string is null or empty"); 795 } 796 797 if (null == anEncodedKey || 0 == anEncodedKey.length()) { 798 throw new IllegalArgumentException( 799 "Illegal argument the encoded key string is null or empty"); 800 } 801 802 // Initialize the member variables with the incoming parameters. 803 this.out = out; 804 805 setKey(aKey); 806 setEncodedKey(anEncodedKey); 807 } 808 809 @Override 810 public synchronized void close() throws IOException { 811 if (out != null) { 812 // Close the output stream and decode the key for the output stream 813 // before returning to the caller. 814 // 815 out.close(); 816 restoreKey(); 817 out = null; 818 } 819 } 820 821 /** 822 * Writes the specified byte to this output stream. The general contract for 823 * write is that one byte is written to the output stream. The byte to be 824 * written is the eight low-order bits of the argument b. The 24 high-order 825 * bits of b are ignored. 826 * 827 * @param b 828 * 32-bit integer of block of 4 bytes 829 */ 830 @Override 831 public void write(int b) throws IOException { 832 out.write(b); 833 } 834 835 /** 836 * Writes b.length bytes from the specified byte array to this output 837 * stream. The general contract for write(b) is that it should have exactly 838 * the same effect as the call write(b, 0, b.length). 839 * 840 * @param b 841 * Block of bytes to be written to the output stream. 842 */ 843 @Override 844 public void write(byte[] b) throws IOException { 845 out.write(b); 846 } 847 848 /** 849 * Writes <code>len</code> from the specified byte array starting at offset 850 * <code>off</code> to the output stream. The general contract for write(b, 851 * off, len) is that some of the bytes in the array <code> 852 * b</code b> are written to the output stream in order; element 853 * <code>b[off]</code> is the first byte written and 854 * <code>b[off+len-1]</code> is the last byte written by this operation. 855 * 856 * @param b 857 * Byte array to be written. 858 * @param off 859 * Write this offset in stream. 860 * @param len 861 * Number of bytes to be written. 862 */ 863 @Override 864 public void write(byte[] b, int off, int len) throws IOException { 865 out.write(b, off, len); 866 } 867 868 /** 869 * Get the blob name. 870 * 871 * @return String Blob name. 872 */ 873 public String getKey() { 874 return key; 875 } 876 877 /** 878 * Set the blob name. 879 * 880 * @param key 881 * Blob name. 882 */ 883 public void setKey(String key) { 884 this.key = key; 885 } 886 887 /** 888 * Get the blob name. 889 * 890 * @return String Blob name. 891 */ 892 public String getEncodedKey() { 893 return keyEncoded; 894 } 895 896 /** 897 * Set the blob name. 898 * 899 * @param anEncodedKey 900 * Blob name. 901 */ 902 public void setEncodedKey(String anEncodedKey) { 903 this.keyEncoded = anEncodedKey; 904 } 905 906 /** 907 * Restore the original key name from the m_key member variable. Note: The 908 * output file stream is created with an encoded blob store key to guarantee 909 * load balancing on the front end of the Azure storage partition servers. 910 * The create also includes the name of the original key value which is 911 * stored in the m_key member variable. This method should only be called 912 * when the stream is closed. 913 * 914 * @param anEncodedKey 915 * Encoding of the original key stored in m_key member. 916 */ 917 private void restoreKey() throws IOException { 918 store.rename(getEncodedKey(), getKey()); 919 } 920 } 921 922 private URI uri; 923 private NativeFileSystemStore store; 924 private AzureNativeFileSystemStore actualStore; 925 private Path workingDir; 926 private long blockSize = MAX_AZURE_BLOCK_SIZE; 927 private AzureFileSystemInstrumentation instrumentation; 928 private String metricsSourceName; 929 private boolean isClosed = false; 930 private static boolean suppressRetryPolicy = false; 931 // A counter to create unique (within-process) names for my metrics sources. 932 private static AtomicInteger metricsSourceNameCounter = new AtomicInteger(); 933 934 935 public NativeAzureFileSystem() { 936 // set store in initialize() 937 } 938 939 public NativeAzureFileSystem(NativeFileSystemStore store) { 940 this.store = store; 941 } 942 943 /** 944 * Suppress the default retry policy for the Storage, useful in unit tests to 945 * test negative cases without waiting forever. 946 */ 947 @VisibleForTesting 948 static void suppressRetryPolicy() { 949 suppressRetryPolicy = true; 950 } 951 952 /** 953 * Undo the effect of suppressRetryPolicy. 954 */ 955 @VisibleForTesting 956 static void resumeRetryPolicy() { 957 suppressRetryPolicy = false; 958 } 959 960 /** 961 * Creates a new metrics source name that's unique within this process. 962 */ 963 @VisibleForTesting 964 public static String newMetricsSourceName() { 965 int number = metricsSourceNameCounter.incrementAndGet(); 966 final String baseName = "AzureFileSystemMetrics"; 967 if (number == 1) { // No need for a suffix for the first one 968 return baseName; 969 } else { 970 return baseName + number; 971 } 972 } 973 974 /** 975 * Checks if the given URI scheme is a scheme that's affiliated with the Azure 976 * File System. 977 * 978 * @param scheme 979 * The URI scheme. 980 * @return true iff it's an Azure File System URI scheme. 981 */ 982 private static boolean isWasbScheme(String scheme) { 983 // The valid schemes are: asv (old name), asvs (old name over HTTPS), 984 // wasb (new name), wasbs (new name over HTTPS). 985 return scheme != null 986 && (scheme.equalsIgnoreCase("asv") || scheme.equalsIgnoreCase("asvs") 987 || scheme.equalsIgnoreCase("wasb") || scheme 988 .equalsIgnoreCase("wasbs")); 989 } 990 991 /** 992 * Puts in the authority of the default file system if it is a WASB file 993 * system and the given URI's authority is null. 994 * 995 * @return The URI with reconstructed authority if necessary and possible. 996 */ 997 private static URI reconstructAuthorityIfNeeded(URI uri, Configuration conf) { 998 if (null == uri.getAuthority()) { 999 // If WASB is the default file system, get the authority from there 1000 URI defaultUri = FileSystem.getDefaultUri(conf); 1001 if (defaultUri != null && isWasbScheme(defaultUri.getScheme())) { 1002 try { 1003 // Reconstruct the URI with the authority from the default URI. 1004 return new URI(uri.getScheme(), defaultUri.getAuthority(), 1005 uri.getPath(), uri.getQuery(), uri.getFragment()); 1006 } catch (URISyntaxException e) { 1007 // This should never happen. 1008 throw new Error("Bad URI construction", e); 1009 } 1010 } 1011 } 1012 return uri; 1013 } 1014 1015 @Override 1016 protected void checkPath(Path path) { 1017 // Make sure to reconstruct the path's authority if needed 1018 super.checkPath(new Path(reconstructAuthorityIfNeeded(path.toUri(), 1019 getConf()))); 1020 } 1021 1022 @Override 1023 public void initialize(URI uri, Configuration conf) 1024 throws IOException, IllegalArgumentException { 1025 // Check authority for the URI to guarantee that it is non-null. 1026 uri = reconstructAuthorityIfNeeded(uri, conf); 1027 if (null == uri.getAuthority()) { 1028 final String errMsg = String 1029 .format("Cannot initialize WASB file system, URI authority not recognized."); 1030 throw new IllegalArgumentException(errMsg); 1031 } 1032 super.initialize(uri, conf); 1033 1034 if (store == null) { 1035 store = createDefaultStore(conf); 1036 } 1037 1038 // Make sure the metrics system is available before interacting with Azure 1039 AzureFileSystemMetricsSystem.fileSystemStarted(); 1040 metricsSourceName = newMetricsSourceName(); 1041 String sourceDesc = "Azure Storage Volume File System metrics"; 1042 instrumentation = new AzureFileSystemInstrumentation(conf); 1043 AzureFileSystemMetricsSystem.registerSource(metricsSourceName, sourceDesc, 1044 instrumentation); 1045 1046 store.initialize(uri, conf, instrumentation); 1047 setConf(conf); 1048 this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority()); 1049 this.workingDir = new Path("/user", UserGroupInformation.getCurrentUser() 1050 .getShortUserName()).makeQualified(getUri(), getWorkingDirectory()); 1051 this.blockSize = conf.getLong(AZURE_BLOCK_SIZE_PROPERTY_NAME, 1052 MAX_AZURE_BLOCK_SIZE); 1053 1054 if (LOG.isDebugEnabled()) { 1055 LOG.debug("NativeAzureFileSystem. Initializing."); 1056 LOG.debug(" blockSize = " 1057 + conf.getLong(AZURE_BLOCK_SIZE_PROPERTY_NAME, MAX_AZURE_BLOCK_SIZE)); 1058 } 1059 } 1060 1061 private NativeFileSystemStore createDefaultStore(Configuration conf) { 1062 actualStore = new AzureNativeFileSystemStore(); 1063 1064 if (suppressRetryPolicy) { 1065 actualStore.suppressRetryPolicy(); 1066 } 1067 return actualStore; 1068 } 1069 1070 /** 1071 * Azure Storage doesn't allow the blob names to end in a period, 1072 * so encode this here to work around that limitation. 1073 */ 1074 private static String encodeTrailingPeriod(String toEncode) { 1075 Matcher matcher = TRAILING_PERIOD_PATTERN.matcher(toEncode); 1076 return matcher.replaceAll(TRAILING_PERIOD_PLACEHOLDER); 1077 } 1078 1079 /** 1080 * Reverse the encoding done by encodeTrailingPeriod(). 1081 */ 1082 private static String decodeTrailingPeriod(String toDecode) { 1083 Matcher matcher = TRAILING_PERIOD_PLACEHOLDER_PATTERN.matcher(toDecode); 1084 return matcher.replaceAll("."); 1085 } 1086 1087 /** 1088 * Convert the path to a key. By convention, any leading or trailing slash is 1089 * removed, except for the special case of a single slash. 1090 */ 1091 @VisibleForTesting 1092 public String pathToKey(Path path) { 1093 // Convert the path to a URI to parse the scheme, the authority, and the 1094 // path from the path object. 1095 URI tmpUri = path.toUri(); 1096 String pathUri = tmpUri.getPath(); 1097 1098 // The scheme and authority is valid. If the path does not exist add a "/" 1099 // separator to list the root of the container. 1100 Path newPath = path; 1101 if ("".equals(pathUri)) { 1102 newPath = new Path(tmpUri.toString() + Path.SEPARATOR); 1103 } 1104 1105 // Verify path is absolute if the path refers to a windows drive scheme. 1106 if (!newPath.isAbsolute()) { 1107 throw new IllegalArgumentException("Path must be absolute: " + path); 1108 } 1109 1110 String key = null; 1111 key = newPath.toUri().getPath(); 1112 key = removeTrailingSlash(key); 1113 key = encodeTrailingPeriod(key); 1114 if (key.length() == 1) { 1115 return key; 1116 } else { 1117 return key.substring(1); // remove initial slash 1118 } 1119 } 1120 1121 // Remove any trailing slash except for the case of a single slash. 1122 private static String removeTrailingSlash(String key) { 1123 if (key.length() == 0 || key.length() == 1) { 1124 return key; 1125 } 1126 if (key.charAt(key.length() - 1) == '/') { 1127 return key.substring(0, key.length() - 1); 1128 } else { 1129 return key; 1130 } 1131 } 1132 1133 private static Path keyToPath(String key) { 1134 if (key.equals("/")) { 1135 return new Path("/"); // container 1136 } 1137 return new Path("/" + decodeTrailingPeriod(key)); 1138 } 1139 1140 /** 1141 * Get the absolute version of the path (fully qualified). 1142 * This is public for testing purposes. 1143 * 1144 * @param path 1145 * @return fully qualified path 1146 */ 1147 @VisibleForTesting 1148 public Path makeAbsolute(Path path) { 1149 if (path.isAbsolute()) { 1150 return path; 1151 } 1152 return new Path(workingDir, path); 1153 } 1154 1155 /** 1156 * For unit test purposes, retrieves the AzureNativeFileSystemStore store 1157 * backing this file system. 1158 * 1159 * @return The store object. 1160 */ 1161 @VisibleForTesting 1162 public AzureNativeFileSystemStore getStore() { 1163 return actualStore; 1164 } 1165 1166 NativeFileSystemStore getStoreInterface() { 1167 return store; 1168 } 1169 1170 /** 1171 * Gets the metrics source for this file system. 1172 * This is mainly here for unit testing purposes. 1173 * 1174 * @return the metrics source. 1175 */ 1176 public AzureFileSystemInstrumentation getInstrumentation() { 1177 return instrumentation; 1178 } 1179 1180 /** This optional operation is not yet supported. */ 1181 @Override 1182 public FSDataOutputStream append(Path f, int bufferSize, Progressable progress) 1183 throws IOException { 1184 throw new IOException("Not supported"); 1185 } 1186 1187 @Override 1188 public FSDataOutputStream create(Path f, FsPermission permission, 1189 boolean overwrite, int bufferSize, short replication, long blockSize, 1190 Progressable progress) throws IOException { 1191 return create(f, permission, overwrite, true, 1192 bufferSize, replication, blockSize, progress, 1193 (SelfRenewingLease) null); 1194 } 1195 1196 /** 1197 * Get a self-renewing lease on the specified file. 1198 */ 1199 public SelfRenewingLease acquireLease(Path path) throws AzureException { 1200 String fullKey = pathToKey(makeAbsolute(path)); 1201 return getStore().acquireLease(fullKey); 1202 } 1203 1204 @Override 1205 @SuppressWarnings("deprecation") 1206 public FSDataOutputStream createNonRecursive(Path f, FsPermission permission, 1207 boolean overwrite, int bufferSize, short replication, long blockSize, 1208 Progressable progress) throws IOException { 1209 1210 Path parent = f.getParent(); 1211 1212 // Get exclusive access to folder if this is a directory designated 1213 // for atomic rename. The primary use case of for HBase write-ahead 1214 // log file management. 1215 SelfRenewingLease lease = null; 1216 if (store.isAtomicRenameKey(pathToKey(f))) { 1217 try { 1218 lease = acquireLease(parent); 1219 } catch (AzureException e) { 1220 1221 String errorCode = ""; 1222 try { 1223 StorageException e2 = (StorageException) e.getCause(); 1224 errorCode = e2.getErrorCode(); 1225 } catch (Exception e3) { 1226 // do nothing if cast fails 1227 } 1228 if (errorCode.equals("BlobNotFound")) { 1229 throw new FileNotFoundException("Cannot create file " + 1230 f.getName() + " because parent folder does not exist."); 1231 } 1232 1233 LOG.warn("Got unexpected exception trying to get lease on " 1234 + pathToKey(parent) + ". " + e.getMessage()); 1235 throw e; 1236 } 1237 } 1238 1239 // See if the parent folder exists. If not, throw error. 1240 // The exists() check will push any pending rename operation forward, 1241 // if there is one, and return false. 1242 // 1243 // At this point, we have exclusive access to the source folder 1244 // via the lease, so we will not conflict with an active folder 1245 // rename operation. 1246 if (!exists(parent)) { 1247 try { 1248 1249 // This'll let the keep-alive thread exit as soon as it wakes up. 1250 lease.free(); 1251 } catch (Exception e) { 1252 LOG.warn("Unable to free lease because: " + e.getMessage()); 1253 } 1254 throw new FileNotFoundException("Cannot create file " + 1255 f.getName() + " because parent folder does not exist."); 1256 } 1257 1258 // Create file inside folder. 1259 FSDataOutputStream out = null; 1260 try { 1261 out = create(f, permission, overwrite, false, 1262 bufferSize, replication, blockSize, progress, lease); 1263 } finally { 1264 // Release exclusive access to folder. 1265 try { 1266 if (lease != null) { 1267 lease.free(); 1268 } 1269 } catch (Exception e) { 1270 IOUtils.cleanup(LOG, out); 1271 String msg = "Unable to free lease on " + parent.toUri(); 1272 LOG.error(msg); 1273 throw new IOException(msg, e); 1274 } 1275 } 1276 return out; 1277 } 1278 1279 @Override 1280 @SuppressWarnings("deprecation") 1281 public FSDataOutputStream createNonRecursive(Path f, FsPermission permission, 1282 EnumSet<CreateFlag> flags, int bufferSize, short replication, long blockSize, 1283 Progressable progress) throws IOException { 1284 1285 // Check if file should be appended or overwritten. Assume that the file 1286 // is overwritten on if the CREATE and OVERWRITE create flags are set. Note 1287 // that any other combinations of create flags will result in an open new or 1288 // open with append. 1289 final EnumSet<CreateFlag> createflags = 1290 EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE); 1291 boolean overwrite = flags.containsAll(createflags); 1292 1293 // Delegate the create non-recursive call. 1294 return this.createNonRecursive(f, permission, overwrite, 1295 bufferSize, replication, blockSize, progress); 1296 } 1297 1298 @Override 1299 @SuppressWarnings("deprecation") 1300 public FSDataOutputStream createNonRecursive(Path f, 1301 boolean overwrite, int bufferSize, short replication, long blockSize, 1302 Progressable progress) throws IOException { 1303 return this.createNonRecursive(f, FsPermission.getFileDefault(), 1304 overwrite, bufferSize, replication, blockSize, progress); 1305 } 1306 1307 1308 /** 1309 * Create an Azure blob and return an output stream to use 1310 * to write data to it. 1311 * 1312 * @param f 1313 * @param permission 1314 * @param overwrite 1315 * @param createParent 1316 * @param bufferSize 1317 * @param replication 1318 * @param blockSize 1319 * @param progress 1320 * @param parentFolderLease Lease on parent folder (or null if 1321 * no lease). 1322 * @return 1323 * @throws IOException 1324 */ 1325 private FSDataOutputStream create(Path f, FsPermission permission, 1326 boolean overwrite, boolean createParent, int bufferSize, 1327 short replication, long blockSize, Progressable progress, 1328 SelfRenewingLease parentFolderLease) 1329 throws IOException { 1330 1331 if (LOG.isDebugEnabled()) { 1332 LOG.debug("Creating file: " + f.toString()); 1333 } 1334 1335 if (containsColon(f)) { 1336 throw new IOException("Cannot create file " + f 1337 + " through WASB that has colons in the name"); 1338 } 1339 1340 Path absolutePath = makeAbsolute(f); 1341 String key = pathToKey(absolutePath); 1342 1343 FileMetadata existingMetadata = store.retrieveMetadata(key); 1344 if (existingMetadata != null) { 1345 if (existingMetadata.isDir()) { 1346 throw new IOException("Cannot create file " + f 1347 + "; already exists as a directory."); 1348 } 1349 if (!overwrite) { 1350 throw new IOException("File already exists:" + f); 1351 } 1352 } 1353 1354 Path parentFolder = absolutePath.getParent(); 1355 if (parentFolder != null && parentFolder.getParent() != null) { // skip root 1356 // Update the parent folder last modified time if the parent folder 1357 // already exists. 1358 String parentKey = pathToKey(parentFolder); 1359 FileMetadata parentMetadata = store.retrieveMetadata(parentKey); 1360 if (parentMetadata != null && parentMetadata.isDir() && 1361 parentMetadata.getBlobMaterialization() == BlobMaterialization.Explicit) { 1362 store.updateFolderLastModifiedTime(parentKey, parentFolderLease); 1363 } else { 1364 // Make sure that the parent folder exists. 1365 // Create it using inherited permissions from the first existing directory going up the path 1366 Path firstExisting = parentFolder.getParent(); 1367 FileMetadata metadata = store.retrieveMetadata(pathToKey(firstExisting)); 1368 while(metadata == null) { 1369 // Guaranteed to terminate properly because we will eventually hit root, which will return non-null metadata 1370 firstExisting = firstExisting.getParent(); 1371 metadata = store.retrieveMetadata(pathToKey(firstExisting)); 1372 } 1373 mkdirs(parentFolder, metadata.getPermissionStatus().getPermission(), true); 1374 } 1375 } 1376 1377 // Mask the permission first (with the default permission mask as well). 1378 FsPermission masked = applyUMask(permission, UMaskApplyMode.NewFile); 1379 PermissionStatus permissionStatus = createPermissionStatus(masked); 1380 1381 OutputStream bufOutStream; 1382 if (store.isPageBlobKey(key)) { 1383 // Store page blobs directly in-place without renames. 1384 bufOutStream = store.storefile(key, permissionStatus); 1385 } else { 1386 // This is a block blob, so open the output blob stream based on the 1387 // encoded key. 1388 // 1389 String keyEncoded = encodeKey(key); 1390 1391 1392 // First create a blob at the real key, pointing back to the temporary file 1393 // This accomplishes a few things: 1394 // 1. Makes sure we can create a file there. 1395 // 2. Makes it visible to other concurrent threads/processes/nodes what 1396 // we're 1397 // doing. 1398 // 3. Makes it easier to restore/cleanup data in the event of us crashing. 1399 store.storeEmptyLinkFile(key, keyEncoded, permissionStatus); 1400 1401 // The key is encoded to point to a common container at the storage server. 1402 // This reduces the number of splits on the server side when load balancing. 1403 // Ingress to Azure storage can take advantage of earlier splits. We remove 1404 // the root path to the key and prefix a random GUID to the tail (or leaf 1405 // filename) of the key. Keys are thus broadly and randomly distributed over 1406 // a single container to ease load balancing on the storage server. When the 1407 // blob is committed it is renamed to its earlier key. Uncommitted blocks 1408 // are not cleaned up and we leave it to Azure storage to garbage collect 1409 // these 1410 // blocks. 1411 bufOutStream = new NativeAzureFsOutputStream(store.storefile( 1412 keyEncoded, permissionStatus), key, keyEncoded); 1413 } 1414 // Construct the data output stream from the buffered output stream. 1415 FSDataOutputStream fsOut = new FSDataOutputStream(bufOutStream, statistics); 1416 1417 1418 // Increment the counter 1419 instrumentation.fileCreated(); 1420 1421 // Return data output stream to caller. 1422 return fsOut; 1423 } 1424 1425 @Override 1426 @Deprecated 1427 public boolean delete(Path path) throws IOException { 1428 return delete(path, true); 1429 } 1430 1431 @Override 1432 public boolean delete(Path f, boolean recursive) throws IOException { 1433 return delete(f, recursive, false); 1434 } 1435 1436 /** 1437 * Delete the specified file or folder. The parameter 1438 * skipParentFolderLastModifidedTimeUpdate 1439 * is used in the case of atomic folder rename redo. In that case, there is 1440 * a lease on the parent folder, so (without reworking the code) modifying 1441 * the parent folder update time will fail because of a conflict with the 1442 * lease. Since we are going to delete the folder soon anyway so accurate 1443 * modified time is not necessary, it's easier to just skip 1444 * the modified time update. 1445 * 1446 * @param f 1447 * @param recursive 1448 * @param skipParentFolderLastModifidedTimeUpdate If true, don't update the folder last 1449 * modified time. 1450 * @return true if and only if the file is deleted 1451 * @throws IOException 1452 */ 1453 public boolean delete(Path f, boolean recursive, 1454 boolean skipParentFolderLastModifidedTimeUpdate) throws IOException { 1455 1456 if (LOG.isDebugEnabled()) { 1457 LOG.debug("Deleting file: " + f.toString()); 1458 } 1459 1460 Path absolutePath = makeAbsolute(f); 1461 String key = pathToKey(absolutePath); 1462 1463 // Capture the metadata for the path. 1464 // 1465 FileMetadata metaFile = store.retrieveMetadata(key); 1466 1467 if (null == metaFile) { 1468 // The path to be deleted does not exist. 1469 return false; 1470 } 1471 1472 // The path exists, determine if it is a folder containing objects, 1473 // an empty folder, or a simple file and take the appropriate actions. 1474 if (!metaFile.isDir()) { 1475 // The path specifies a file. We need to check the parent path 1476 // to make sure it's a proper materialized directory before we 1477 // delete the file. Otherwise we may get into a situation where 1478 // the file we were deleting was the last one in an implicit directory 1479 // (e.g. the blob store only contains the blob a/b and there's no 1480 // corresponding directory blob a) and that would implicitly delete 1481 // the directory as well, which is not correct. 1482 Path parentPath = absolutePath.getParent(); 1483 if (parentPath.getParent() != null) {// Not root 1484 String parentKey = pathToKey(parentPath); 1485 FileMetadata parentMetadata = store.retrieveMetadata(parentKey); 1486 if (!parentMetadata.isDir()) { 1487 // Invalid state: the parent path is actually a file. Throw. 1488 throw new AzureException("File " + f + " has a parent directory " 1489 + parentPath + " which is also a file. Can't resolve."); 1490 } 1491 if (parentMetadata.getBlobMaterialization() == BlobMaterialization.Implicit) { 1492 if (LOG.isDebugEnabled()) { 1493 LOG.debug("Found an implicit parent directory while trying to" 1494 + " delete the file " + f + ". Creating the directory blob for" 1495 + " it in " + parentKey + "."); 1496 } 1497 store.storeEmptyFolder(parentKey, 1498 createPermissionStatus(FsPermission.getDefault())); 1499 } else { 1500 if (!skipParentFolderLastModifidedTimeUpdate) { 1501 store.updateFolderLastModifiedTime(parentKey, null); 1502 } 1503 } 1504 } 1505 store.delete(key); 1506 instrumentation.fileDeleted(); 1507 } else { 1508 // The path specifies a folder. Recursively delete all entries under the 1509 // folder. 1510 Path parentPath = absolutePath.getParent(); 1511 if (parentPath.getParent() != null) { 1512 String parentKey = pathToKey(parentPath); 1513 FileMetadata parentMetadata = store.retrieveMetadata(parentKey); 1514 1515 if (parentMetadata.getBlobMaterialization() == BlobMaterialization.Implicit) { 1516 if (LOG.isDebugEnabled()) { 1517 LOG.debug("Found an implicit parent directory while trying to" 1518 + " delete the directory " + f 1519 + ". Creating the directory blob for" + " it in " + parentKey 1520 + "."); 1521 } 1522 store.storeEmptyFolder(parentKey, 1523 createPermissionStatus(FsPermission.getDefault())); 1524 } 1525 } 1526 1527 // List all the blobs in the current folder. 1528 String priorLastKey = null; 1529 PartialListing listing = store.listAll(key, AZURE_LIST_ALL, 1, 1530 priorLastKey); 1531 FileMetadata[] contents = listing.getFiles(); 1532 if (!recursive && contents.length > 0) { 1533 // The folder is non-empty and recursive delete was not specified. 1534 // Throw an exception indicating that a non-recursive delete was 1535 // specified for a non-empty folder. 1536 throw new IOException("Non-recursive delete of non-empty directory " 1537 + f.toString()); 1538 } 1539 1540 // Delete all the files in the folder. 1541 for (FileMetadata p : contents) { 1542 // Tag on the directory name found as the suffix of the suffix of the 1543 // parent directory to get the new absolute path. 1544 String suffix = p.getKey().substring( 1545 p.getKey().lastIndexOf(PATH_DELIMITER)); 1546 if (!p.isDir()) { 1547 store.delete(key + suffix); 1548 instrumentation.fileDeleted(); 1549 } else { 1550 // Recursively delete contents of the sub-folders. Notice this also 1551 // deletes the blob for the directory. 1552 if (!delete(new Path(f.toString() + suffix), true)) { 1553 return false; 1554 } 1555 } 1556 } 1557 store.delete(key); 1558 1559 // Update parent directory last modified time 1560 Path parent = absolutePath.getParent(); 1561 if (parent != null && parent.getParent() != null) { // not root 1562 String parentKey = pathToKey(parent); 1563 if (!skipParentFolderLastModifidedTimeUpdate) { 1564 store.updateFolderLastModifiedTime(parentKey, null); 1565 } 1566 } 1567 instrumentation.directoryDeleted(); 1568 } 1569 1570 // File or directory was successfully deleted. 1571 return true; 1572 } 1573 1574 @Override 1575 public FileStatus getFileStatus(Path f) throws IOException { 1576 1577 if (LOG.isDebugEnabled()) { 1578 LOG.debug("Getting the file status for " + f.toString()); 1579 } 1580 1581 // Capture the absolute path and the path to key. 1582 Path absolutePath = makeAbsolute(f); 1583 String key = pathToKey(absolutePath); 1584 if (key.length() == 0) { // root always exists 1585 return newDirectory(null, absolutePath); 1586 } 1587 1588 // The path is either a folder or a file. Retrieve metadata to 1589 // determine if it is a directory or file. 1590 FileMetadata meta = store.retrieveMetadata(key); 1591 if (meta != null) { 1592 if (meta.isDir()) { 1593 // The path is a folder with files in it. 1594 // 1595 if (LOG.isDebugEnabled()) { 1596 LOG.debug("Path " + f.toString() + "is a folder."); 1597 } 1598 1599 // If a rename operation for the folder was pending, redo it. 1600 // Then the file does not exist, so signal that. 1601 if (conditionalRedoFolderRename(f)) { 1602 throw new FileNotFoundException( 1603 absolutePath + ": No such file or directory."); 1604 } 1605 1606 // Return reference to the directory object. 1607 return newDirectory(meta, absolutePath); 1608 } 1609 1610 // The path is a file. 1611 if (LOG.isDebugEnabled()) { 1612 LOG.debug("Found the path: " + f.toString() + " as a file."); 1613 } 1614 1615 // Return with reference to a file object. 1616 return newFile(meta, absolutePath); 1617 } 1618 1619 // File not found. Throw exception no such file or directory. 1620 // 1621 throw new FileNotFoundException( 1622 absolutePath + ": No such file or directory."); 1623 } 1624 1625 // Return true if there is a rename pending and we redo it, otherwise false. 1626 private boolean conditionalRedoFolderRename(Path f) throws IOException { 1627 1628 // Can't rename /, so return immediately in that case. 1629 if (f.getName().equals("")) { 1630 return false; 1631 } 1632 1633 // Check if there is a -RenamePending.json file for this folder, and if so, 1634 // redo the rename. 1635 Path absoluteRenamePendingFile = renamePendingFilePath(f); 1636 if (exists(absoluteRenamePendingFile)) { 1637 FolderRenamePending pending = 1638 new FolderRenamePending(absoluteRenamePendingFile, this); 1639 pending.redo(); 1640 return true; 1641 } else { 1642 return false; 1643 } 1644 } 1645 1646 // Return the path name that would be used for rename of folder with path f. 1647 private Path renamePendingFilePath(Path f) { 1648 Path absPath = makeAbsolute(f); 1649 String key = pathToKey(absPath); 1650 key += "-RenamePending.json"; 1651 return keyToPath(key); 1652 } 1653 1654 @Override 1655 public URI getUri() { 1656 return uri; 1657 } 1658 1659 /** 1660 * Retrieve the status of a given path if it is a file, or of all the 1661 * contained files if it is a directory. 1662 */ 1663 @Override 1664 public FileStatus[] listStatus(Path f) throws IOException { 1665 1666 if (LOG.isDebugEnabled()) { 1667 LOG.debug("Listing status for " + f.toString()); 1668 } 1669 1670 Path absolutePath = makeAbsolute(f); 1671 String key = pathToKey(absolutePath); 1672 Set<FileStatus> status = new TreeSet<FileStatus>(); 1673 FileMetadata meta = store.retrieveMetadata(key); 1674 1675 if (meta != null) { 1676 if (!meta.isDir()) { 1677 if (LOG.isDebugEnabled()) { 1678 LOG.debug("Found path as a file"); 1679 } 1680 return new FileStatus[] { newFile(meta, absolutePath) }; 1681 } 1682 String partialKey = null; 1683 PartialListing listing = store.list(key, AZURE_LIST_ALL, 1, partialKey); 1684 1685 // For any -RenamePending.json files in the listing, 1686 // push the rename forward. 1687 boolean renamed = conditionalRedoFolderRenames(listing); 1688 1689 // If any renames were redone, get another listing, 1690 // since the current one may have changed due to the redo. 1691 if (renamed) { 1692 listing = store.list(key, AZURE_LIST_ALL, 1, partialKey); 1693 } 1694 1695 for (FileMetadata fileMetadata : listing.getFiles()) { 1696 Path subpath = keyToPath(fileMetadata.getKey()); 1697 1698 // Test whether the metadata represents a file or directory and 1699 // add the appropriate metadata object. 1700 // 1701 // Note: There was a very old bug here where directories were added 1702 // to the status set as files flattening out recursive listings 1703 // using "-lsr" down the file system hierarchy. 1704 if (fileMetadata.isDir()) { 1705 // Make sure we hide the temp upload folder 1706 if (fileMetadata.getKey().equals(AZURE_TEMP_FOLDER)) { 1707 // Don't expose that. 1708 continue; 1709 } 1710 status.add(newDirectory(fileMetadata, subpath)); 1711 } else { 1712 status.add(newFile(fileMetadata, subpath)); 1713 } 1714 } 1715 if (LOG.isDebugEnabled()) { 1716 LOG.debug("Found path as a directory with " + status.size() 1717 + " files in it."); 1718 } 1719 } else { 1720 // There is no metadata found for the path. 1721 if (LOG.isDebugEnabled()) { 1722 LOG.debug("Did not find any metadata for path: " + key); 1723 } 1724 1725 throw new FileNotFoundException("File" + f + " does not exist."); 1726 } 1727 1728 return status.toArray(new FileStatus[0]); 1729 } 1730 1731 // Redo any folder renames needed if there are rename pending files in the 1732 // directory listing. Return true if one or more redo operations were done. 1733 private boolean conditionalRedoFolderRenames(PartialListing listing) 1734 throws IllegalArgumentException, IOException { 1735 boolean renamed = false; 1736 for (FileMetadata fileMetadata : listing.getFiles()) { 1737 Path subpath = keyToPath(fileMetadata.getKey()); 1738 if (isRenamePendingFile(subpath)) { 1739 FolderRenamePending pending = 1740 new FolderRenamePending(subpath, this); 1741 pending.redo(); 1742 renamed = true; 1743 } 1744 } 1745 return renamed; 1746 } 1747 1748 // True if this is a folder rename pending file, else false. 1749 private boolean isRenamePendingFile(Path path) { 1750 return path.toString().endsWith(FolderRenamePending.SUFFIX); 1751 } 1752 1753 private FileStatus newFile(FileMetadata meta, Path path) { 1754 return new FileStatus ( 1755 meta.getLength(), 1756 false, 1757 1, 1758 blockSize, 1759 meta.getLastModified(), 1760 0, 1761 meta.getPermissionStatus().getPermission(), 1762 meta.getPermissionStatus().getUserName(), 1763 meta.getPermissionStatus().getGroupName(), 1764 path.makeQualified(getUri(), getWorkingDirectory())); 1765 } 1766 1767 private FileStatus newDirectory(FileMetadata meta, Path path) { 1768 return new FileStatus ( 1769 0, 1770 true, 1771 1, 1772 blockSize, 1773 meta == null ? 0 : meta.getLastModified(), 1774 0, 1775 meta == null ? FsPermission.getDefault() : meta.getPermissionStatus().getPermission(), 1776 meta == null ? "" : meta.getPermissionStatus().getUserName(), 1777 meta == null ? "" : meta.getPermissionStatus().getGroupName(), 1778 path.makeQualified(getUri(), getWorkingDirectory())); 1779 } 1780 1781 private static enum UMaskApplyMode { 1782 NewFile, 1783 NewDirectory, 1784 NewDirectoryNoUmask, 1785 ChangeExistingFile, 1786 ChangeExistingDirectory, 1787 } 1788 1789 /** 1790 * Applies the applicable UMASK's on the given permission. 1791 * 1792 * @param permission 1793 * The permission to mask. 1794 * @param applyDefaultUmask 1795 * Whether to also apply the default umask. 1796 * @return The masked persmission. 1797 */ 1798 private FsPermission applyUMask(final FsPermission permission, 1799 final UMaskApplyMode applyMode) { 1800 FsPermission newPermission = new FsPermission(permission); 1801 // Apply the default umask - this applies for new files or directories. 1802 if (applyMode == UMaskApplyMode.NewFile 1803 || applyMode == UMaskApplyMode.NewDirectory) { 1804 newPermission = newPermission 1805 .applyUMask(FsPermission.getUMask(getConf())); 1806 } 1807 return newPermission; 1808 } 1809 1810 /** 1811 * Creates the PermissionStatus object to use for the given permission, based 1812 * on the current user in context. 1813 * 1814 * @param permission 1815 * The permission for the file. 1816 * @return The permission status object to use. 1817 * @throws IOException 1818 * If login fails in getCurrentUser 1819 */ 1820 private PermissionStatus createPermissionStatus(FsPermission permission) 1821 throws IOException { 1822 // Create the permission status for this file based on current user 1823 return new PermissionStatus( 1824 UserGroupInformation.getCurrentUser().getShortUserName(), 1825 getConf().get(AZURE_DEFAULT_GROUP_PROPERTY_NAME, 1826 AZURE_DEFAULT_GROUP_DEFAULT), 1827 permission); 1828 } 1829 1830 @Override 1831 public boolean mkdirs(Path f, FsPermission permission) throws IOException { 1832 return mkdirs(f, permission, false); 1833 } 1834 1835 public boolean mkdirs(Path f, FsPermission permission, boolean noUmask) throws IOException { 1836 if (LOG.isDebugEnabled()) { 1837 LOG.debug("Creating directory: " + f.toString()); 1838 } 1839 1840 if (containsColon(f)) { 1841 throw new IOException("Cannot create directory " + f 1842 + " through WASB that has colons in the name"); 1843 } 1844 1845 Path absolutePath = makeAbsolute(f); 1846 PermissionStatus permissionStatus = null; 1847 if(noUmask) { 1848 // ensure owner still has wx permissions at the minimum 1849 permissionStatus = createPermissionStatus( 1850 applyUMask(FsPermission.createImmutable((short) (permission.toShort() | USER_WX_PERMISION)), 1851 UMaskApplyMode.NewDirectoryNoUmask)); 1852 } else { 1853 permissionStatus = createPermissionStatus( 1854 applyUMask(permission, UMaskApplyMode.NewDirectory)); 1855 } 1856 1857 1858 ArrayList<String> keysToCreateAsFolder = new ArrayList<String>(); 1859 ArrayList<String> keysToUpdateAsFolder = new ArrayList<String>(); 1860 boolean childCreated = false; 1861 // Check that there is no file in the parent chain of the given path. 1862 for (Path current = absolutePath, parent = current.getParent(); 1863 parent != null; // Stop when you get to the root 1864 current = parent, parent = current.getParent()) { 1865 String currentKey = pathToKey(current); 1866 FileMetadata currentMetadata = store.retrieveMetadata(currentKey); 1867 if (currentMetadata != null && !currentMetadata.isDir()) { 1868 throw new IOException("Cannot create directory " + f + " because " + 1869 current + " is an existing file."); 1870 } else if (currentMetadata == null) { 1871 keysToCreateAsFolder.add(currentKey); 1872 childCreated = true; 1873 } else { 1874 // The directory already exists. Its last modified time need to be 1875 // updated if there is a child directory created under it. 1876 if (childCreated) { 1877 keysToUpdateAsFolder.add(currentKey); 1878 } 1879 childCreated = false; 1880 } 1881 } 1882 1883 for (String currentKey : keysToCreateAsFolder) { 1884 store.storeEmptyFolder(currentKey, permissionStatus); 1885 } 1886 1887 instrumentation.directoryCreated(); 1888 1889 // otherwise throws exception 1890 return true; 1891 } 1892 1893 @Override 1894 public FSDataInputStream open(Path f, int bufferSize) throws IOException { 1895 if (LOG.isDebugEnabled()) { 1896 LOG.debug("Opening file: " + f.toString()); 1897 } 1898 1899 Path absolutePath = makeAbsolute(f); 1900 String key = pathToKey(absolutePath); 1901 FileMetadata meta = store.retrieveMetadata(key); 1902 if (meta == null) { 1903 throw new FileNotFoundException(f.toString()); 1904 } 1905 if (meta.isDir()) { 1906 throw new FileNotFoundException(f.toString() 1907 + " is a directory not a file."); 1908 } 1909 1910 return new FSDataInputStream(new BufferedFSInputStream( 1911 new NativeAzureFsInputStream(store.retrieve(key), key, meta.getLength()), bufferSize)); 1912 } 1913 1914 @Override 1915 public boolean rename(Path src, Path dst) throws IOException { 1916 1917 FolderRenamePending renamePending = null; 1918 1919 if (LOG.isDebugEnabled()) { 1920 LOG.debug("Moving " + src + " to " + dst); 1921 } 1922 1923 if (containsColon(dst)) { 1924 throw new IOException("Cannot rename to file " + dst 1925 + " through WASB that has colons in the name"); 1926 } 1927 1928 String srcKey = pathToKey(makeAbsolute(src)); 1929 1930 if (srcKey.length() == 0) { 1931 // Cannot rename root of file system 1932 return false; 1933 } 1934 1935 // Figure out the final destination 1936 Path absoluteDst = makeAbsolute(dst); 1937 String dstKey = pathToKey(absoluteDst); 1938 FileMetadata dstMetadata = store.retrieveMetadata(dstKey); 1939 if (dstMetadata != null && dstMetadata.isDir()) { 1940 // It's an existing directory. 1941 dstKey = pathToKey(makeAbsolute(new Path(dst, src.getName()))); 1942 if (LOG.isDebugEnabled()) { 1943 LOG.debug("Destination " + dst 1944 + " is a directory, adjusted the destination to be " + dstKey); 1945 } 1946 } else if (dstMetadata != null) { 1947 // Attempting to overwrite a file using rename() 1948 if (LOG.isDebugEnabled()) { 1949 LOG.debug("Destination " + dst 1950 + " is an already existing file, failing the rename."); 1951 } 1952 return false; 1953 } else { 1954 // Check that the parent directory exists. 1955 FileMetadata parentOfDestMetadata = 1956 store.retrieveMetadata(pathToKey(absoluteDst.getParent())); 1957 if (parentOfDestMetadata == null) { 1958 if (LOG.isDebugEnabled()) { 1959 LOG.debug("Parent of the destination " + dst 1960 + " doesn't exist, failing the rename."); 1961 } 1962 return false; 1963 } else if (!parentOfDestMetadata.isDir()) { 1964 if (LOG.isDebugEnabled()) { 1965 LOG.debug("Parent of the destination " + dst 1966 + " is a file, failing the rename."); 1967 } 1968 return false; 1969 } 1970 } 1971 FileMetadata srcMetadata = store.retrieveMetadata(srcKey); 1972 if (srcMetadata == null) { 1973 // Source doesn't exist 1974 if (LOG.isDebugEnabled()) { 1975 LOG.debug("Source " + src + " doesn't exist, failing the rename."); 1976 } 1977 return false; 1978 } else if (!srcMetadata.isDir()) { 1979 if (LOG.isDebugEnabled()) { 1980 LOG.debug("Source " + src + " found as a file, renaming."); 1981 } 1982 store.rename(srcKey, dstKey); 1983 } else { 1984 1985 // Prepare for, execute and clean up after of all files in folder, and 1986 // the root file, and update the last modified time of the source and 1987 // target parent folders. The operation can be redone if it fails part 1988 // way through, by applying the "Rename Pending" file. 1989 1990 // The following code (internally) only does atomic rename preparation 1991 // and lease management for page blob folders, limiting the scope of the 1992 // operation to HBase log file folders, where atomic rename is required. 1993 // In the future, we could generalize it easily to all folders. 1994 renamePending = prepareAtomicFolderRename(srcKey, dstKey); 1995 renamePending.execute(); 1996 if (LOG.isDebugEnabled()) { 1997 LOG.debug("Renamed " + src + " to " + dst + " successfully."); 1998 } 1999 renamePending.cleanup(); 2000 return true; 2001 } 2002 2003 // Update the last-modified time of the parent folders of both source 2004 // and destination. 2005 updateParentFolderLastModifiedTime(srcKey); 2006 updateParentFolderLastModifiedTime(dstKey); 2007 2008 if (LOG.isDebugEnabled()) { 2009 LOG.debug("Renamed " + src + " to " + dst + " successfully."); 2010 } 2011 return true; 2012 } 2013 2014 /** 2015 * Update the last-modified time of the parent folder of the file 2016 * identified by key. 2017 * @param key 2018 * @throws IOException 2019 */ 2020 private void updateParentFolderLastModifiedTime(String key) 2021 throws IOException { 2022 Path parent = makeAbsolute(keyToPath(key)).getParent(); 2023 if (parent != null && parent.getParent() != null) { // not root 2024 String parentKey = pathToKey(parent); 2025 2026 // ensure the parent is a materialized folder 2027 FileMetadata parentMetadata = store.retrieveMetadata(parentKey); 2028 // The metadata could be null if the implicit folder only contains a 2029 // single file. In this case, the parent folder no longer exists if the 2030 // file is renamed; so we can safely ignore the null pointer case. 2031 if (parentMetadata != null) { 2032 if (parentMetadata.isDir() 2033 && parentMetadata.getBlobMaterialization() == BlobMaterialization.Implicit) { 2034 store.storeEmptyFolder(parentKey, 2035 createPermissionStatus(FsPermission.getDefault())); 2036 } 2037 2038 store.updateFolderLastModifiedTime(parentKey, null); 2039 } 2040 } 2041 } 2042 2043 /** 2044 * If the source is a page blob folder, 2045 * prepare to rename this folder atomically. This means to get exclusive 2046 * access to the source folder, and record the actions to be performed for 2047 * this rename in a "Rename Pending" file. This code was designed to 2048 * meet the needs of HBase, which requires atomic rename of write-ahead log 2049 * (WAL) folders for correctness. 2050 * 2051 * Before calling this method, the caller must ensure that the source is a 2052 * folder. 2053 * 2054 * For non-page-blob directories, prepare the in-memory information needed, 2055 * but don't take the lease or write the redo file. This is done to limit the 2056 * scope of atomic folder rename to HBase, at least at the time of writing 2057 * this code. 2058 * 2059 * @param srcKey Source folder name. 2060 * @param dstKey Destination folder name. 2061 * @throws IOException 2062 */ 2063 private FolderRenamePending prepareAtomicFolderRename( 2064 String srcKey, String dstKey) throws IOException { 2065 2066 if (store.isAtomicRenameKey(srcKey)) { 2067 2068 // Block unwanted concurrent access to source folder. 2069 SelfRenewingLease lease = leaseSourceFolder(srcKey); 2070 2071 // Prepare in-memory information needed to do or redo a folder rename. 2072 FolderRenamePending renamePending = 2073 new FolderRenamePending(srcKey, dstKey, lease, this); 2074 2075 // Save it to persistent storage to help recover if the operation fails. 2076 renamePending.writeFile(this); 2077 return renamePending; 2078 } else { 2079 FolderRenamePending renamePending = 2080 new FolderRenamePending(srcKey, dstKey, null, this); 2081 return renamePending; 2082 } 2083 } 2084 2085 /** 2086 * Get a self-renewing Azure blob lease on the source folder zero-byte file. 2087 */ 2088 private SelfRenewingLease leaseSourceFolder(String srcKey) 2089 throws AzureException { 2090 return store.acquireLease(srcKey); 2091 } 2092 2093 /** 2094 * Return an array containing hostnames, offset and size of 2095 * portions of the given file. For WASB we'll just lie and give 2096 * fake hosts to make sure we get many splits in MR jobs. 2097 */ 2098 @Override 2099 public BlockLocation[] getFileBlockLocations(FileStatus file, 2100 long start, long len) throws IOException { 2101 if (file == null) { 2102 return null; 2103 } 2104 2105 if ((start < 0) || (len < 0)) { 2106 throw new IllegalArgumentException("Invalid start or len parameter"); 2107 } 2108 2109 if (file.getLen() < start) { 2110 return new BlockLocation[0]; 2111 } 2112 final String blobLocationHost = getConf().get( 2113 AZURE_BLOCK_LOCATION_HOST_PROPERTY_NAME, 2114 AZURE_BLOCK_LOCATION_HOST_DEFAULT); 2115 final String[] name = { blobLocationHost }; 2116 final String[] host = { blobLocationHost }; 2117 long blockSize = file.getBlockSize(); 2118 if (blockSize <= 0) { 2119 throw new IllegalArgumentException( 2120 "The block size for the given file is not a positive number: " 2121 + blockSize); 2122 } 2123 int numberOfLocations = (int) (len / blockSize) 2124 + ((len % blockSize == 0) ? 0 : 1); 2125 BlockLocation[] locations = new BlockLocation[numberOfLocations]; 2126 for (int i = 0; i < locations.length; i++) { 2127 long currentOffset = start + (i * blockSize); 2128 long currentLength = Math.min(blockSize, start + len - currentOffset); 2129 locations[i] = new BlockLocation(name, host, currentOffset, currentLength); 2130 } 2131 return locations; 2132 } 2133 2134 /** 2135 * Set the working directory to the given directory. 2136 */ 2137 @Override 2138 public void setWorkingDirectory(Path newDir) { 2139 workingDir = makeAbsolute(newDir); 2140 } 2141 2142 @Override 2143 public Path getWorkingDirectory() { 2144 return workingDir; 2145 } 2146 2147 @Override 2148 public void setPermission(Path p, FsPermission permission) throws IOException { 2149 Path absolutePath = makeAbsolute(p); 2150 String key = pathToKey(absolutePath); 2151 FileMetadata metadata = store.retrieveMetadata(key); 2152 if (metadata == null) { 2153 throw new FileNotFoundException("File doesn't exist: " + p); 2154 } 2155 permission = applyUMask(permission, 2156 metadata.isDir() ? UMaskApplyMode.ChangeExistingDirectory 2157 : UMaskApplyMode.ChangeExistingFile); 2158 if (metadata.getBlobMaterialization() == BlobMaterialization.Implicit) { 2159 // It's an implicit folder, need to materialize it. 2160 store.storeEmptyFolder(key, createPermissionStatus(permission)); 2161 } else if (!metadata.getPermissionStatus().getPermission(). 2162 equals(permission)) { 2163 store.changePermissionStatus(key, new PermissionStatus( 2164 metadata.getPermissionStatus().getUserName(), 2165 metadata.getPermissionStatus().getGroupName(), 2166 permission)); 2167 } 2168 } 2169 2170 @Override 2171 public void setOwner(Path p, String username, String groupname) 2172 throws IOException { 2173 Path absolutePath = makeAbsolute(p); 2174 String key = pathToKey(absolutePath); 2175 FileMetadata metadata = store.retrieveMetadata(key); 2176 if (metadata == null) { 2177 throw new FileNotFoundException("File doesn't exist: " + p); 2178 } 2179 PermissionStatus newPermissionStatus = new PermissionStatus( 2180 username == null ? 2181 metadata.getPermissionStatus().getUserName() : username, 2182 groupname == null ? 2183 metadata.getPermissionStatus().getGroupName() : groupname, 2184 metadata.getPermissionStatus().getPermission()); 2185 if (metadata.getBlobMaterialization() == BlobMaterialization.Implicit) { 2186 // It's an implicit folder, need to materialize it. 2187 store.storeEmptyFolder(key, newPermissionStatus); 2188 } else { 2189 store.changePermissionStatus(key, newPermissionStatus); 2190 } 2191 } 2192 2193 @Override 2194 public synchronized void close() throws IOException { 2195 if (isClosed) { 2196 return; 2197 } 2198 2199 // Call the base close() to close any resources there. 2200 super.close(); 2201 // Close the store to close any resources there - e.g. the bandwidth 2202 // updater thread would be stopped at this time. 2203 store.close(); 2204 // Notify the metrics system that this file system is closed, which may 2205 // trigger one final metrics push to get the accurate final file system 2206 // metrics out. 2207 2208 long startTime = System.currentTimeMillis(); 2209 2210 AzureFileSystemMetricsSystem.unregisterSource(metricsSourceName); 2211 AzureFileSystemMetricsSystem.fileSystemClosed(); 2212 2213 if (LOG.isDebugEnabled()) { 2214 LOG.debug("Submitting metrics when file system closed took " 2215 + (System.currentTimeMillis() - startTime) + " ms."); 2216 } 2217 isClosed = true; 2218 } 2219 2220 /** 2221 * A handler that defines what to do with blobs whose upload was 2222 * interrupted. 2223 */ 2224 private abstract class DanglingFileHandler { 2225 abstract void handleFile(FileMetadata file, FileMetadata tempFile) 2226 throws IOException; 2227 } 2228 2229 /** 2230 * Handler implementation for just deleting dangling files and cleaning 2231 * them up. 2232 */ 2233 private class DanglingFileDeleter extends DanglingFileHandler { 2234 @Override 2235 void handleFile(FileMetadata file, FileMetadata tempFile) 2236 throws IOException { 2237 if (LOG.isDebugEnabled()) { 2238 LOG.debug("Deleting dangling file " + file.getKey()); 2239 } 2240 store.delete(file.getKey()); 2241 store.delete(tempFile.getKey()); 2242 } 2243 } 2244 2245 /** 2246 * Handler implementation for just moving dangling files to recovery 2247 * location (/lost+found). 2248 */ 2249 private class DanglingFileRecoverer extends DanglingFileHandler { 2250 private final Path destination; 2251 2252 DanglingFileRecoverer(Path destination) { 2253 this.destination = destination; 2254 } 2255 2256 @Override 2257 void handleFile(FileMetadata file, FileMetadata tempFile) 2258 throws IOException { 2259 if (LOG.isDebugEnabled()) { 2260 LOG.debug("Recovering " + file.getKey()); 2261 } 2262 // Move to the final destination 2263 String finalDestinationKey = 2264 pathToKey(new Path(destination, file.getKey())); 2265 store.rename(tempFile.getKey(), finalDestinationKey); 2266 if (!finalDestinationKey.equals(file.getKey())) { 2267 // Delete the empty link file now that we've restored it. 2268 store.delete(file.getKey()); 2269 } 2270 } 2271 } 2272 2273 /** 2274 * Check if a path has colons in its name 2275 */ 2276 private boolean containsColon(Path p) { 2277 return p.toUri().getPath().toString().contains(":"); 2278 } 2279 2280 /** 2281 * Implements recover and delete (-move and -delete) behaviors for handling 2282 * dangling files (blobs whose upload was interrupted). 2283 * 2284 * @param root 2285 * The root path to check from. 2286 * @param handler 2287 * The handler that deals with dangling files. 2288 */ 2289 private void handleFilesWithDanglingTempData(Path root, 2290 DanglingFileHandler handler) throws IOException { 2291 // Calculate the cut-off for when to consider a blob to be dangling. 2292 long cutoffForDangling = new Date().getTime() 2293 - getConf().getInt(AZURE_TEMP_EXPIRY_PROPERTY_NAME, 2294 AZURE_TEMP_EXPIRY_DEFAULT) * 1000; 2295 // Go over all the blobs under the given root and look for blobs to 2296 // recover. 2297 String priorLastKey = null; 2298 do { 2299 PartialListing listing = store.listAll(pathToKey(root), AZURE_LIST_ALL, 2300 AZURE_UNBOUNDED_DEPTH, priorLastKey); 2301 2302 for (FileMetadata file : listing.getFiles()) { 2303 if (!file.isDir()) { // We don't recover directory blobs 2304 // See if this blob has a link in it (meaning it's a place-holder 2305 // blob for when the upload to the temp blob is complete). 2306 String link = store.getLinkInFileMetadata(file.getKey()); 2307 if (link != null) { 2308 // It has a link, see if the temp blob it is pointing to is 2309 // existent and old enough to be considered dangling. 2310 FileMetadata linkMetadata = store.retrieveMetadata(link); 2311 if (linkMetadata != null 2312 && linkMetadata.getLastModified() >= cutoffForDangling) { 2313 // Found one! 2314 handler.handleFile(file, linkMetadata); 2315 } 2316 } 2317 } 2318 } 2319 priorLastKey = listing.getPriorLastKey(); 2320 } while (priorLastKey != null); 2321 } 2322 2323 /** 2324 * Looks under the given root path for any blob that are left "dangling", 2325 * meaning that they are place-holder blobs that we created while we upload 2326 * the data to a temporary blob, but for some reason we crashed in the middle 2327 * of the upload and left them there. If any are found, we move them to the 2328 * destination given. 2329 * 2330 * @param root 2331 * The root path to consider. 2332 * @param destination 2333 * The destination path to move any recovered files to. 2334 * @throws IOException 2335 */ 2336 public void recoverFilesWithDanglingTempData(Path root, Path destination) 2337 throws IOException { 2338 if (LOG.isDebugEnabled()) { 2339 LOG.debug("Recovering files with dangling temp data in " + root); 2340 } 2341 handleFilesWithDanglingTempData(root, 2342 new DanglingFileRecoverer(destination)); 2343 } 2344 2345 /** 2346 * Looks under the given root path for any blob that are left "dangling", 2347 * meaning that they are place-holder blobs that we created while we upload 2348 * the data to a temporary blob, but for some reason we crashed in the middle 2349 * of the upload and left them there. If any are found, we delete them. 2350 * 2351 * @param root 2352 * The root path to consider. 2353 * @throws IOException 2354 */ 2355 public void deleteFilesWithDanglingTempData(Path root) throws IOException { 2356 if (LOG.isDebugEnabled()) { 2357 LOG.debug("Deleting files with dangling temp data in " + root); 2358 } 2359 handleFilesWithDanglingTempData(root, new DanglingFileDeleter()); 2360 } 2361 2362 @Override 2363 protected void finalize() throws Throwable { 2364 LOG.debug("finalize() called."); 2365 close(); 2366 super.finalize(); 2367 } 2368 2369 /** 2370 * Encode the key with a random prefix for load balancing in Azure storage. 2371 * Upload data to a random temporary file then do storage side renaming to 2372 * recover the original key. 2373 * 2374 * @param aKey 2375 * @param numBuckets 2376 * @return Encoded version of the original key. 2377 */ 2378 private static String encodeKey(String aKey) { 2379 // Get the tail end of the key name. 2380 // 2381 String fileName = aKey.substring(aKey.lastIndexOf(Path.SEPARATOR) + 1, 2382 aKey.length()); 2383 2384 // Construct the randomized prefix of the file name. The prefix ensures the 2385 // file always drops into the same folder but with a varying tail key name. 2386 String filePrefix = AZURE_TEMP_FOLDER + Path.SEPARATOR 2387 + UUID.randomUUID().toString(); 2388 2389 // Concatenate the randomized prefix with the tail of the key name. 2390 String randomizedKey = filePrefix + fileName; 2391 2392 // Return to the caller with the randomized key. 2393 return randomizedKey; 2394 } 2395}