Merge pull request #2157 from swordqiu/feature/qj-bucket-usages

feature: bucket usage and quota check

cmd/climc/shell/buckets.go

@@ -233,10 +233,15 @@ func init() {
 	})
 
 	type BucketSyncOptions struct {
-		ID string `help:"ID or name of bucket" json:"-"`
+		ID        string `help:"ID or name of bucket" json:"-"`
+		StatsOnly bool   `help:"sync statistics only"`
 	}
 	R(&BucketSyncOptions{}, "bucket-sync", "Sync bucket", func(s *mcclient.ClientSession, args *BucketSyncOptions) error {
-		result, err := modules.Buckets.PerformAction(s, args.ID, "sync", nil)
+		params, err := options.StructToParams(args)
+		if err != nil {
+			return err
+		}
+		result, err := modules.Buckets.PerformAction(s, args.ID, "sync", params)
 		if err != nil {
 			return err
 		}

pkg/cloudcommon/db/project_resources.go

@@ -66,7 +66,8 @@ type SProjectResourceCount struct {
 }
 
 func (virtman *SVirtualResourceBaseManager) GetResourceCount() ([]SProjectResourceCount, error) {
-	virts := virtman.Query()
+	virts := virtman.GetIVirtualModelManager().Query()
+	// log.Debugf("GetResourceCount: %s", virtman.keywordPlural)
 	return CalculateProjectResourceCount(virts)
 }
 

pkg/cloudprovider/objectstore.go

@@ -200,9 +200,12 @@ func GetIObjects(bucket ICloudBucket, objectPrefix string, isRecursive bool) ([]
 		}
 
 		// Send all objects
-		if len(result.Objects) > 0 {
-			ret = append(ret, result.Objects...)
-			marker = result.Objects[len(result.Objects)-1].GetKey()
+		for i := range result.Objects {
+			if !isRecursive && result.Objects[i].GetKey() == objectPrefix {
+				continue
+			}
+			ret = append(ret, result.Objects[i])
+			marker = result.Objects[i].GetKey()
 		}
 
 		// Send all common prefixes if any.

pkg/compute/models/buckets.go

@@ -39,6 +39,7 @@ import (
 	"yunion.io/x/onecloud/pkg/httperrors"
 	"yunion.io/x/onecloud/pkg/mcclient"
 	"yunion.io/x/onecloud/pkg/mcclient/modules"
+	"yunion.io/x/onecloud/pkg/util/rbacutils"
 )
 
 type SBucketManager struct {
@@ -379,6 +380,21 @@ func (manager *SBucketManager) ValidateCreateData(
 	if err != nil {
 		return nil, httperrors.NewInputParameterError("invalid bucket name: %s", err)
 	}
+
+	managerId, _ := data.GetString("manager_id")
+	if len(managerId) == 0 {
+		return nil, httperrors.NewInternalServerError("empty manager_id???")
+	}
+	cloudprovider := CloudproviderManager.FetchCloudproviderById(managerId)
+	if cloudprovider == nil {
+		return nil, httperrors.NewInternalServerError("invalid cloudprovider???")
+	}
+	quotaPlatformId := cloudprovider.GetQuotaPlatformID()
+	pendingUsage := SQuota{Bucket: 1}
+	if err := QuotaManager.CheckSetPendingQuota(ctx, userCred, rbacutils.ScopeProject, ownerId, quotaPlatformId, &pendingUsage); err != nil {
+		return nil, httperrors.NewOutOfQuotaError("%s", err)
+	}
+
 	return manager.SVirtualResourceBaseManager.ValidateCreateData(ctx, userCred, ownerId, query, data)
 }
 
@@ -389,6 +405,14 @@ func (bucket *SBucket) PostCreate(
 	query jsonutils.JSONObject,
 	data jsonutils.JSONObject,
 ) {
+	cloudprovider := bucket.GetCloudprovider()
+	quotaPlatformId := cloudprovider.GetQuotaPlatformID()
+	pendingUsage := SQuota{Bucket: 1}
+	err := QuotaManager.CancelPendingUsage(ctx, userCred, rbacutils.ScopeProject, ownerId, quotaPlatformId, &pendingUsage, &pendingUsage)
+	if err != nil {
+		log.Errorf("CancelPendingUsage error %s", err)
+	}
+
 	bucket.SetStatus(userCred, api.BUCKET_STATUS_START_CREATE, "PostCreate")
 	task, err := taskman.TaskManager.NewTask(ctx, "BucketCreateTask", bucket, userCred, nil, "", "", nil)
 	if err != nil {
@@ -769,12 +793,14 @@ func (bucket *SBucket) PerformSync(
 	query jsonutils.JSONObject,
 	data jsonutils.JSONObject,
 ) (jsonutils.JSONObject, error) {
+	statsOnly := jsonutils.QueryBoolean(data, "stats_only", false)
+
 	iBucket, err := bucket.GetIBucket()
 	if err != nil {
 		return nil, httperrors.NewInternalServerError("fail to find external bucket: %s", err)
 	}
 
-	err = bucket.syncWithCloudBucket(ctx, userCred, iBucket, nil, false)
+	err = bucket.syncWithCloudBucket(ctx, userCred, iBucket, nil, statsOnly)
 	if err != nil {
 		return nil, httperrors.NewInternalServerError("syncWithCloudBucket error %s", err)
 	}
@@ -828,3 +854,74 @@ func (bucket *SBucket) GetDetailsAcl(
 	ret.Add(jsonutils.NewString(string(acl)), "acl")
 	return ret, nil
 }
+
+func (manager *SBucketManager) usageQByCloudEnv(q *sqlchemy.SQuery, providers []string, brands []string, cloudEnv string) *sqlchemy.SQuery {
+	return CloudProviderFilter(q, q.Field("manager_id"), providers, brands, cloudEnv)
+}
+
+func (manager *SBucketManager) usageQByRange(q *sqlchemy.SQuery, rangeObj db.IStandaloneModel) *sqlchemy.SQuery {
+	if rangeObj == nil {
+		return q
+	}
+
+	kw := rangeObj.Keyword()
+	switch kw {
+	case "zone":
+		zone := rangeObj.(*SZone)
+		q = q.Filter(sqlchemy.Equals(q.Field("cloudregion_id"), zone.CloudregionId))
+	case "wire":
+		wire := rangeObj.(*SWire)
+		zone := wire.GetZone()
+		q = q.Filter(sqlchemy.Equals(q.Field("cloudregion_id"), zone.CloudregionId))
+	case "host":
+		host := rangeObj.(*SHost)
+		zone := host.GetZone()
+		q = q.Filter(sqlchemy.Equals(q.Field("cloudregion_id"), zone.CloudregionId))
+	case "cloudprovider":
+		q = q.Filter(sqlchemy.Equals(q.Field("manager_id"), rangeObj.GetId()))
+	case "cloudaccount":
+		cloudproviders := CloudproviderManager.Query().SubQuery()
+		subq := cloudproviders.Query(cloudproviders.Field("id")).Equals("cloudaccount_id", rangeObj.GetId()).SubQuery()
+		q = q.Filter(sqlchemy.In(q.Field("manager_id"), subq))
+	case "cloudregion":
+		q = q.Filter(sqlchemy.Equals(q.Field("cloudregion_id"), rangeObj.GetId()))
+	}
+
+	return q
+}
+
+func (manager *SBucketManager) usageQ(q *sqlchemy.SQuery, rangeObj db.IStandaloneModel, providers []string, brands []string, cloudEnv string) *sqlchemy.SQuery {
+	q = manager.usageQByRange(q, rangeObj)
+	q = manager.usageQByCloudEnv(q, providers, brands, cloudEnv)
+	return q
+}
+
+type SBucketUsages struct {
+	Buckets int
+	Objects int
+	Bytes   int64
+}
+
+func (manager *SBucketManager) TotalCount(scope rbacutils.TRbacScope, ownerId mcclient.IIdentityProvider, rangeObj db.IStandaloneModel, providers []string, brands []string, cloudEnv string) SBucketUsages {
+	usage := SBucketUsages{}
+	buckets := manager.Query().SubQuery()
+	q := buckets.Query(
+		sqlchemy.COUNT("buckets"),
+		sqlchemy.SUM("objects", buckets.Field("object_cnt")),
+		sqlchemy.SUM("bytes", buckets.Field("size_bytes")),
+	)
+	q = manager.usageQ(q, rangeObj, providers, brands, cloudEnv)
+	switch scope {
+	case rbacutils.ScopeSystem:
+		// do nothing
+	case rbacutils.ScopeDomain:
+		q = q.Equals("domain_id", ownerId.GetProjectDomainId())
+	case rbacutils.ScopeProject:
+		q = q.Equals("tenant_id", ownerId.GetProjectId())
+	}
+	err := q.First(&usage)
+	if err != nil {
+		log.Errorf("Query bucket usage error %s", err)
+	}
+	return usage
+}

pkg/compute/models/quotas.go

@@ -66,6 +66,8 @@ type SQuota struct {
 	Secgroup       int
 	IsolatedDevice int
 	Snapshot       int
+
+	Bucket int
 }
 
 func (self *SQuota) FetchSystemQuota(scope rbacutils.TRbacScope, ownerId mcclient.IIdentityProvider) {
@@ -87,6 +89,7 @@ func (self *SQuota) FetchSystemQuota(scope rbacutils.TRbacScope, ownerId mcclien
 	self.Secgroup = options.Options.DefaultSecgroupQuota * base
 	self.IsolatedDevice = options.Options.DefaultIsolatedDeviceQuota * base
 	self.Snapshot = options.Options.DefaultSnapshotQuota * base
+	self.Bucket = options.Options.DefaultBucketQuota * base
 }
 
 func (self *SQuota) FetchUsage(ctx context.Context, scope rbacutils.TRbacScope, ownerId mcclient.IIdentityProvider, name []string) error {
@@ -99,6 +102,7 @@ func (self *SQuota) FetchUsage(ctx context.Context, scope rbacutils.TRbacScope,
 	guest := totalGuestResourceCount(scope, ownerId, nil, nil, hypervisors.List(), false, false, nil, nil, nil, nil, "")
 	eipUsage := ElasticipManager.TotalCount(scope, ownerId, nil, nil, nil, "")
 	snapshotCount, _ := TotalSnapshotCount(scope, ownerId, nil, nil, nil, "")
+	bucketUsage := BucketManager.TotalCount(scope, ownerId, nil, nil, nil, "")
 	// XXX
 	// keypair belongs to user
 	// keypair := totalKeypairCount(projectId)
@@ -120,6 +124,7 @@ func (self *SQuota) FetchUsage(ctx context.Context, scope rbacutils.TRbacScope,
 	self.Secgroup, _ = totalSecurityGroupCount(scope, ownerId)
 	self.IsolatedDevice = guest.TotalIsolatedCount
 	self.Snapshot = snapshotCount
+	self.Bucket = bucketUsage.Buckets
 	return nil
 }
 
@@ -160,6 +165,9 @@ func (self *SQuota) IsEmpty() bool {
 	if self.Snapshot > 0 {
 		return false
 	}
+	if self.Bucket > 0 {
+		return false
+	}
 	return true
 }
 
@@ -177,6 +185,7 @@ func (self *SQuota) Add(quota quotas.IQuota) {
 	self.Secgroup = self.Secgroup + squota.Secgroup
 	self.IsolatedDevice = self.IsolatedDevice + squota.IsolatedDevice
 	self.Snapshot = self.Snapshot + squota.Snapshot
+	self.Bucket = self.Bucket + squota.Bucket
 }
 
 func nonNegative(val int) int {
@@ -197,6 +206,7 @@ func (self *SQuota) Sub(quota quotas.IQuota) {
 	self.Secgroup = nonNegative(self.Secgroup - squota.Secgroup)
 	self.IsolatedDevice = nonNegative(self.IsolatedDevice - squota.IsolatedDevice)
 	self.Snapshot = nonNegative(self.Snapshot - squota.Snapshot)
+	self.Bucket = nonNegative(self.Bucket - squota.Bucket)
 }
 
 func (self *SQuota) Update(quota quotas.IQuota) {
@@ -237,6 +247,9 @@ func (self *SQuota) Update(quota quotas.IQuota) {
 	if squota.Snapshot > 0 {
 		self.Snapshot = squota.Snapshot
 	}
+	if squota.Bucket > 0 {
+		self.Bucket = squota.Bucket
+	}
 }
 
 func (self *SQuota) Exceed(request quotas.IQuota, quota quotas.IQuota) error {
@@ -279,6 +292,9 @@ func (self *SQuota) Exceed(request quotas.IQuota, quota quotas.IQuota) error {
 	if sreq.Snapshot > 0 && self.Snapshot > squota.Snapshot {
 		err.Add("snapshot", squota.Snapshot, self.Snapshot)
 	}
+	if sreq.Bucket > 0 && self.Bucket > squota.Bucket {
+		err.Add("bucket", squota.Bucket, self.Bucket)
+	}
 	if err.IsError() {
 		return err
 	} else {
@@ -296,41 +312,18 @@ func keyName(prefix, name string) string {
 
 func (self *SQuota) ToJSON(prefix string) jsonutils.JSONObject {
 	ret := jsonutils.NewDict()
-	// if self.Cpu > 0 {
 	ret.Add(jsonutils.NewInt(int64(self.Cpu)), keyName(prefix, "cpu"))
-	//}
-	// if self.Memory > 0 {
 	ret.Add(jsonutils.NewInt(int64(self.Memory)), keyName(prefix, "memory"))
-	//}
-	//if self.Storage > 0 {
 	ret.Add(jsonutils.NewInt(int64(self.Storage)), keyName(prefix, "storage"))
-	//}
-	//if self.Port > 0 {
 	ret.Add(jsonutils.NewInt(int64(self.Port)), keyName(prefix, "port"))
-	//}
-	//if self.Eip > 0 {
 	ret.Add(jsonutils.NewInt(int64(self.Eip)), keyName(prefix, "eip"))
-	//}
-	//if self.Eport > 0 {
 	ret.Add(jsonutils.NewInt(int64(self.Eport)), keyName(prefix, "eport"))
-	//}
-	//if self.Bw > 0 {
 	ret.Add(jsonutils.NewInt(int64(self.Bw)), keyName(prefix, "bw"))
-	//}
-	//if self.Ebw > 0 {
 	ret.Add(jsonutils.NewInt(int64(self.Ebw)), keyName(prefix, "ebw"))
-	//}
-	//if self.Group > 0 {
 	ret.Add(jsonutils.NewInt(int64(self.Group)), keyName(prefix, "group"))
-	//}
-	//if self.Secgroup > 0 {
 	ret.Add(jsonutils.NewInt(int64(self.Secgroup)), keyName(prefix, "secgroup"))
-	//}
-	//if self.IsolatedDevice > 0 {
 	ret.Add(jsonutils.NewInt(int64(self.IsolatedDevice)), keyName(prefix, "isolated_device"))
-	//}
-	//if self.Snapshot > 0 {
 	ret.Add(jsonutils.NewInt(int64(self.Snapshot)), keyName(prefix, "snapshot"))
-	//}
+	ret.Add(jsonutils.NewInt(int64(self.Bucket)), keyName(prefix, "bucket"))
 	return ret
 }

pkg/compute/options/options.go

@@ -63,6 +63,7 @@ type ComputeOptions struct {
 	DefaultSecgroupQuota       int `default:"50" help:"Common security group quota per tenant, default 50"`
 	DefaultIsolatedDeviceQuota int `default:"200" help:"Common isolated device quota per tenant, default 200"`
 	DefaultSnapshotQuota       int `default:"10" help:"Common snapshot quota per tenant, default 10"`
+	DefaultBucketQuota         int `default:"100" help:"Common bucket quota per tenant, default 100"`
 
 	SystemAdminQuotaCheck bool `help:"Enable quota check for system admin, default False" default:"false"`
 

pkg/compute/usages/handler.go

@@ -278,6 +278,8 @@ func getAdminGeneralUsage(userCred mcclient.IIdentityProvider, rangeObj db.IStan
 
 		EipUsage(rbacutils.ScopeSystem, nil, rangeObj, providers, brands, cloudEnv),
 
+		BucketUsage(rbacutils.ScopeSystem, nil, rangeObj, providers, brands, cloudEnv),
+
 		SnapshotUsage(rbacutils.ScopeSystem, nil, rangeObj, providers, brands, cloudEnv),
 	)
 
@@ -291,6 +293,8 @@ func getCommonGeneralUsage(scope rbacutils.TRbacScope, cred mcclient.IIdentityPr
 
 	eipUsage := EipUsage(scope, cred, rangeObj, providers, brands, cloudEnv)
 
+	bucketUsage := BucketUsage(scope, cred, rangeObj, providers, brands, cloudEnv)
+
 	snapshotUsage := SnapshotUsage(scope, cred, rangeObj, providers, brands, cloudEnv)
 
 	disksUsage := disksUsage("", rangeObj, nil, nil, providers, brands, cloudEnv, scope, cred)
@@ -319,6 +323,8 @@ func getCommonGeneralUsage(scope rbacutils.TRbacScope, cred mcclient.IIdentityPr
 
 		eipUsage,
 
+		bucketUsage,
+
 		snapshotUsage,
 
 		disksUsage,
@@ -609,6 +615,16 @@ func EipUsage(scope rbacutils.TRbacScope, ownerId mcclient.IIdentityProvider, ra
 	return count
 }
 
+func BucketUsage(scope rbacutils.TRbacScope, ownerId mcclient.IIdentityProvider, rangeObj db.IStandaloneModel, providers []string, brands []string, cloudEnv string) Usage {
+	projectId := mcclient.OwnerIdString(ownerId, scope)
+	bucketUsage := models.BucketManager.TotalCount(scope, ownerId, rangeObj, providers, brands, cloudEnv)
+	count := make(map[string]interface{})
+	count[getKey(projectId, "buckets")] = bucketUsage.Buckets
+	count[getKey(projectId, "bucket_objects")] = bucketUsage.Objects
+	count[getKey(projectId, "bucket_bytes")] = bucketUsage.Bytes
+	return count
+}
+
 func SnapshotUsage(scope rbacutils.TRbacScope, ownerId mcclient.IIdentityProvider, rangeObj db.IStandaloneModel, providers []string, brands []string, cloudEnv string) Usage {
 	projectId := mcclient.OwnerIdString(ownerId, scope)
 	cnt, _ := models.TotalSnapshotCount(scope, ownerId, rangeObj, providers, brands, cloudEnv)

pkg/multicloud/objectstore/buckets.go

@@ -130,6 +130,9 @@ func (bucket *SBucket) GetIObjects(prefix string, isRecursive bool) ([]cloudprov
 		if object.Err != nil {
 			return nil, errors.Wrap(object.Err, "ListObjects")
 		}
+		if !isRecursive && prefix == object.Key {
+			continue
+		}
 		obj := &SObject{
 			bucket: bucket,
 			SBaseCloudObject: cloudprovider.SBaseCloudObject{