HDFS-4091. Add snapshot quota to limit the number of snapshots allowed.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/HDFS-2802@1401868 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES.HDFS-2802.txt

@@ -23,3 +23,6 @@ Branch-2802 Snapshot (Unreleased)
 
   HDFS-4084. Provide CLI support to allow and disallow snapshot 
   on a directory. (Brondon Li via suresh)
+
+  HDFS-4091. Add snapshot quota to limit the number of snapshots allowed.
+  (szetszwo)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/INodeDirectorySnapshottable.java

@@ -21,6 +21,7 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
+import org.apache.hadoop.HadoopIllegalArgumentException;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.hdfs.server.namenode.INode;
 import org.apache.hadoop.hdfs.server.namenode.INodeDirectory;
@@ -30,7 +31,8 @@ import org.apache.hadoop.util.Time;
 /** Directories where taking snapshots is allowed. */
 @InterfaceAudience.Private
 public class INodeDirectorySnapshottable extends INodeDirectoryWithQuota {
-  static public INodeDirectorySnapshottable newInstance(final INodeDirectory dir) {
+  static public INodeDirectorySnapshottable newInstance(
+      final INodeDirectory dir, final int snapshotQuota) {
     long nsq = -1L;
     long dsq = -1L;
 
@@ -39,7 +41,7 @@ public class INodeDirectorySnapshottable extends INodeDirectoryWithQuota {
       nsq = q.getNsQuota();
       dsq = q.getDsQuota();
     }
-    return new INodeDirectorySnapshottable(nsq, dsq, dir);
+    return new INodeDirectorySnapshottable(nsq, dsq, dir, snapshotQuota);
   }
 
   /** Cast INode to INodeDirectorySnapshottable. */
@@ -55,10 +57,25 @@ public class INodeDirectorySnapshottable extends INodeDirectoryWithQuota {
   /** A list of snapshots of this directory. */
   private final List<INodeDirectorySnapshotRoot> snapshots
       = new ArrayList<INodeDirectorySnapshotRoot>();
+  /** Number of snapshots is allowed. */
+  private int snapshotQuota;
 
   private INodeDirectorySnapshottable(long nsQuota, long dsQuota,
-      INodeDirectory dir) {
+      INodeDirectory dir, final int snapshotQuota) {
     super(nsQuota, dsQuota, dir);
+    setSnapshotQuota(snapshotQuota);
+  }
+
+  public int getSnapshotQuota() {
+    return snapshotQuota;
+  }
+
+  public void setSnapshotQuota(int snapshotQuota) {
+    if (snapshotQuota <= 0) {
+      throw new HadoopIllegalArgumentException(
+          "Cannot set snapshot quota to " + snapshotQuota + " <= 0");
+    }
+    this.snapshotQuota = snapshotQuota;
   }
 
   @Override
@@ -67,7 +84,15 @@ public class INodeDirectorySnapshottable extends INodeDirectoryWithQuota {
   }
 
   /** Add a snapshot root under this directory. */
-  INodeDirectorySnapshotRoot addSnapshotRoot(final String name) {
+  INodeDirectorySnapshotRoot addSnapshotRoot(final String name
+      ) throws SnapshotException {
+    //check snapshot quota
+    if (snapshots.size() + 1 > snapshotQuota) {
+      throw new SnapshotException("Failed to add snapshot: there are already "
+          + snapshots.size() + " snapshot(s) and the snapshot quota is "
+          + snapshotQuota);
+    }
+
     final INodeDirectorySnapshotRoot r = new INodeDirectorySnapshotRoot(name, this);
     snapshots.add(r);
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/SnapshotManager.java

@@ -43,7 +43,7 @@ public class SnapshotManager {
    * Otherwise, the {@link INodeDirectory} of the path is replaced by an 
    * {@link INodeDirectorySnapshottable}.
    */
-  public void setSnapshottable(final String path,
+  public void setSnapshottable(final String path, final int snapshotQuota,
       final FSDirectory fsdir) throws IOException {
     namesystem.writeLock();
     try {
@@ -54,7 +54,7 @@ public class SnapshotManager {
       }
 
       final INodeDirectorySnapshottable s
-          = INodeDirectorySnapshottable.newInstance(d);
+          = INodeDirectorySnapshottable.newInstance(d, snapshotQuota);
       fsdir.replaceINodeDirectory(path, d, s);
       snapshottables.add(s);
     } finally {