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}