QPID-4268 C++ Broker Acl support for limiting on-disk store file size and count

Reviewed at https://reviews.apache.org/r/6838/



git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1382095 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 63 insertions, 1 deletions

diff --git a/cpp/src/qpid/acl/AclData.cpp b/cpp/src/qpid/acl/AclData.cpp
index ac97e53b14..997fbdf2eb 100644
--- a/cpp/src/qpid/acl/AclData.cpp
+++ b/cpp/src/qpid/acl/AclData.cpp
@@ -199,6 +199,16 @@ namespace acl {
                                     lookupParamItr = params->find(PROP_MAXQUEUESIZE);
                                     break;
 
+                                case acl::SPECPROP_MAXFILECOUNTUPPERLIMIT:
+                                case acl::SPECPROP_MAXFILECOUNTLOWERLIMIT:
+                                    lookupParamItr = params->find(PROP_MAXFILECOUNT);
+                                    break;
+
+                                case acl::SPECPROP_MAXFILESIZEUPPERLIMIT:
+                                case acl::SPECPROP_MAXFILESIZELOWERLIMIT:
+                                    lookupParamItr = params->find(PROP_MAXFILESIZE);
+                                    break;
+
                                 default:
                                     lookupParamItr = params->find((Property)rulePropMapItr->first);
                                     break;
@@ -222,6 +232,8 @@ namespace acl {
                                     {
                                     case acl::SPECPROP_MAXQUEUECOUNTUPPERLIMIT:
                                     case acl::SPECPROP_MAXQUEUESIZEUPPERLIMIT:
+                                    case acl::SPECPROP_MAXFILECOUNTUPPERLIMIT:
+                                    case acl::SPECPROP_MAXFILESIZEUPPERLIMIT:
                                         limitChecked &=
                                             compareIntMax(
                                                 rulePropMapItr->first,
@@ -231,6 +243,8 @@ namespace acl {
 
                                     case acl::SPECPROP_MAXQUEUECOUNTLOWERLIMIT:
                                     case acl::SPECPROP_MAXQUEUESIZELOWERLIMIT:
+                                    case acl::SPECPROP_MAXFILECOUNTLOWERLIMIT:
+                                    case acl::SPECPROP_MAXFILESIZELOWERLIMIT:
                                         limitChecked &=
                                             compareIntMin(
                                                 rulePropMapItr->first,
diff --git a/cpp/src/qpid/acl/AclValidator.cpp b/cpp/src/qpid/acl/AclValidator.cpp
index 85f0f7c240..73b49b2959 100644
--- a/cpp/src/qpid/acl/AclValidator.cpp
+++ b/cpp/src/qpid/acl/AclValidator.cpp
@@ -94,6 +94,22 @@ namespace acl {
                                     boost::shared_ptr<PropertyType>(
                                         new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
 
+        validators.insert(Validator(acl::SPECPROP_MAXFILESIZELOWERLIMIT,
+                                    boost::shared_ptr<PropertyType>(
+                                        new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
+        validators.insert(Validator(acl::SPECPROP_MAXFILESIZEUPPERLIMIT,
+                                    boost::shared_ptr<PropertyType>(
+                                        new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
+        validators.insert(Validator(acl::SPECPROP_MAXFILECOUNTLOWERLIMIT,
+                                    boost::shared_ptr<PropertyType>(
+                                        new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
+        validators.insert(Validator(acl::SPECPROP_MAXFILECOUNTUPPERLIMIT,
+                                    boost::shared_ptr<PropertyType>(
+                                        new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
         std::string policyTypes[] = {"ring", "ring_strict", "flow_to_disk", "reject"};
         std::vector<std::string> v(policyTypes, policyTypes + sizeof(policyTypes) / sizeof(std::string));
         validators.insert(Validator(acl::SPECPROP_POLICYTYPE,
diff --git a/cpp/src/qpid/broker/AclModule.h b/cpp/src/qpid/broker/AclModule.h
index 987d3e4a78..4070e49559 100644
--- a/cpp/src/qpid/broker/AclModule.h
+++ b/cpp/src/qpid/broker/AclModule.h
@@ -79,6 +79,8 @@ namespace acl {
         PROP_POLICYTYPE,
         PROP_MAXQUEUESIZE,
         PROP_MAXQUEUECOUNT,
+        PROP_MAXFILESIZE,
+        PROP_MAXFILECOUNT,
         PROPERTYSIZE           // PROPERTYSIZE must be last in list
     };
 
@@ -102,7 +104,11 @@ namespace acl {
         SPECPROP_MAXQUEUESIZELOWERLIMIT,
         SPECPROP_MAXQUEUESIZEUPPERLIMIT,
         SPECPROP_MAXQUEUECOUNTLOWERLIMIT,
-        SPECPROP_MAXQUEUECOUNTUPPERLIMIT };
+        SPECPROP_MAXQUEUECOUNTUPPERLIMIT,
+        SPECPROP_MAXFILESIZELOWERLIMIT,
+        SPECPROP_MAXFILESIZEUPPERLIMIT,
+        SPECPROP_MAXFILECOUNTLOWERLIMIT,
+        SPECPROP_MAXFILECOUNTUPPERLIMIT };
 
 // AclResult  shared between ACL spec and ACL authorise interface
     enum AclResult {
@@ -227,6 +233,8 @@ namespace acl {
             if (str.compare("policytype")    == 0) return PROP_POLICYTYPE;
             if (str.compare("maxqueuesize")  == 0) return PROP_MAXQUEUESIZE;
             if (str.compare("maxqueuecount") == 0) return PROP_MAXQUEUECOUNT;
+            if (str.compare("maxfilesize")   == 0) return PROP_MAXFILESIZE;
+            if (str.compare("maxfilecount")  == 0) return PROP_MAXFILECOUNT;
             throw qpid::Exception(str);
         }
         static inline std::string getPropertyStr(const Property p) {
@@ -245,6 +253,8 @@ namespace acl {
             case PROP_POLICYTYPE:    return "policytype";
             case PROP_MAXQUEUESIZE:  return "maxqueuesize";
             case PROP_MAXQUEUECOUNT: return "maxqueuecount";
+            case PROP_MAXFILESIZE:   return "maxfilesize";
+            case PROP_MAXFILECOUNT:  return "maxfilecount";
             default: assert(false); // should never get here
             }
             return "";
@@ -266,6 +276,10 @@ namespace acl {
             if (str.compare("queuemaxsizeupperlimit")   == 0) return SPECPROP_MAXQUEUESIZEUPPERLIMIT;
             if (str.compare("queuemaxcountlowerlimit")  == 0) return SPECPROP_MAXQUEUECOUNTLOWERLIMIT;
             if (str.compare("queuemaxcountupperlimit")  == 0) return SPECPROP_MAXQUEUECOUNTUPPERLIMIT;
+            if (str.compare("filemaxsizelowerlimit")    == 0) return SPECPROP_MAXFILESIZELOWERLIMIT;
+            if (str.compare("filemaxsizeupperlimit")    == 0) return SPECPROP_MAXFILESIZEUPPERLIMIT;
+            if (str.compare("filemaxcountlowerlimit")   == 0) return SPECPROP_MAXFILECOUNTLOWERLIMIT;
+            if (str.compare("filemaxcountupperlimit")   == 0) return SPECPROP_MAXFILECOUNTUPPERLIMIT;
             // Allow old names in ACL file as aliases for newly-named properties
             if (str.compare("maxqueuesize")             == 0) return SPECPROP_MAXQUEUESIZEUPPERLIMIT;
             if (str.compare("maxqueuecount")            == 0) return SPECPROP_MAXQUEUECOUNTUPPERLIMIT;
@@ -289,6 +303,10 @@ namespace acl {
                 case SPECPROP_MAXQUEUESIZEUPPERLIMIT:  return "queuemaxsizeupperlimit";
                 case SPECPROP_MAXQUEUECOUNTLOWERLIMIT: return "queuemaxcountlowerlimit";
                 case SPECPROP_MAXQUEUECOUNTUPPERLIMIT: return "queuemaxcountupperlimit";
+                case SPECPROP_MAXFILESIZELOWERLIMIT:   return "filemaxsizelowerlimit";
+                case SPECPROP_MAXFILESIZEUPPERLIMIT:   return "filemaxsizeupperlimit";
+                case SPECPROP_MAXFILECOUNTLOWERLIMIT:  return "filemaxcountlowerlimit";
+                case SPECPROP_MAXFILECOUNTUPPERLIMIT:  return "filemaxcountupperlimit";
                 default: assert(false); // should never get here
             }
             return "";
diff --git a/cpp/src/qpid/broker/Broker.cpp b/cpp/src/qpid/broker/Broker.cpp
index c67d06f50c..979ad1c45e 100644
--- a/cpp/src/qpid/broker/Broker.cpp
+++ b/cpp/src/qpid/broker/Broker.cpp
@@ -1069,6 +1069,8 @@ std::pair<boost::shared_ptr<Queue>, bool> Broker::createQueue(
         params.insert(make_pair(acl::PROP_POLICYTYPE, settings.dropMessagesAtLimit ? "ring" : "reject"));
         params.insert(make_pair(acl::PROP_MAXQUEUECOUNT, boost::lexical_cast<string>(settings.maxDepth.getCount())));
         params.insert(make_pair(acl::PROP_MAXQUEUESIZE, boost::lexical_cast<string>(settings.maxDepth.getSize())));
+        params.insert(make_pair(acl::PROP_MAXFILECOUNT, boost::lexical_cast<string>(settings.maxFileCount)));
+        params.insert(make_pair(acl::PROP_MAXFILESIZE, boost::lexical_cast<string>(settings.maxFileSize)));
 
         if (!acl->authorise(userId,acl::ACT_CREATE,acl::OBJ_QUEUE,name,&params) )
             throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied queue create request from " << userId));
diff --git a/cpp/src/qpid/broker/QueueSettings.cpp b/cpp/src/qpid/broker/QueueSettings.cpp
index 91616636f1..b180a89e43 100644
--- a/cpp/src/qpid/broker/QueueSettings.cpp
+++ b/cpp/src/qpid/broker/QueueSettings.cpp
@@ -33,6 +33,8 @@ namespace broker {
 namespace {
 const std::string MAX_COUNT("qpid.max_count");
 const std::string MAX_SIZE("qpid.max_size");
+const std::string MAX_FILE_COUNT("qpid.file_count");
+const std::string MAX_FILE_SIZE("qpid.file_size");
 const std::string POLICY_TYPE("qpid.policy_type");
 const std::string POLICY_TYPE_REJECT("reject");
 const std::string POLICY_TYPE_RING("ring");
@@ -163,6 +165,12 @@ bool QueueSettings::handle(const std::string& key, const qpid::types::Variant& v
     } else if (key == ALERT_SIZE) {
         alertThreshold.setSize(value);
         return true;
+    } else if (key == MAX_FILE_COUNT && value.asUint64() > 0) {
+        maxFileCount = value.asUint64();
+        return false; // 'handle' here and also pass to store
+    } else if (key == MAX_FILE_SIZE && value.asUint64() > 0) {
+        maxFileSize = value.asUint64();
+        return false; // 'handle' here and also pass to store
     } else {
         return false;
     }
diff --git a/cpp/src/qpid/broker/QueueSettings.h b/cpp/src/qpid/broker/QueueSettings.h
index 2443624615..3256266df7 100644
--- a/cpp/src/qpid/broker/QueueSettings.h
+++ b/cpp/src/qpid/broker/QueueSettings.h
@@ -71,6 +71,10 @@ struct QueueSettings
     QueueDepth alertThreshold;
     int64_t alertRepeatInterval;
 
+    //file limits checked by Acl and shared with storeSettings
+    uint64_t maxFileSize;
+    uint64_t maxFileCount;
+
     //yuck, yuck
     qpid::framing::FieldTable storeSettings;
     std::map<std::string, qpid::types::Variant> original;