Storage: limit the number of uploaded files (#50796)

* #50608: sql file upload quotas

* rename `files_in_sql` to `file`

* merge conflict

pkg/services/quota/quotaimpl/quota.go

@@ -190,6 +190,11 @@ func (s *Service) getQuotaScopes(target string) ([]models.QuotaScope, error) {
 			models.QuotaScope{Name: "org", Target: target, DefaultLimit: s.Cfg.Quota.Org.AlertRule},
 		)
 		return scopes, nil
+	case "file":
+		scopes = append(scopes,
+			models.QuotaScope{Name: "global", Target: target, DefaultLimit: s.Cfg.Quota.Global.File},
+		)
+		return scopes, nil
 	default:
 		return scopes, quota.ErrInvalidQuotaTarget
 	}

pkg/services/sqlstore/quota.go

@@ -12,6 +12,7 @@ import (
 const (
 	alertRuleTarget = "alert_rule"
 	dashboardTarget = "dashboard"
+	filesTarget     = "file"
 )
 
 type targetCount struct {
@@ -255,7 +256,19 @@ func (ss *SQLStore) UpdateUserQuota(ctx context.Context, cmd *models.UpdateUserQ
 func (ss *SQLStore) GetGlobalQuotaByTarget(ctx context.Context, query *models.GetGlobalQuotaByTargetQuery) error {
 	return ss.WithDbSession(ctx, func(sess *DBSession) error {
 		var used int64
-		if query.Target != alertRuleTarget || query.UnifiedAlertingEnabled {
+
+		if query.Target == filesTarget {
+			// get quota used.
+			rawSQL := fmt.Sprintf("SELECT COUNT(*) AS count FROM %s",
+				dialect.Quote("file"))
+
+			notFolderCondition := fmt.Sprintf(" WHERE path NOT LIKE '%s'", "%/")
+			resp := make([]*targetCount, 0)
+			if err := sess.SQL(rawSQL + notFolderCondition).Find(&resp); err != nil {
+				return err
+			}
+			used = resp[0].Count
+		} else if query.Target != alertRuleTarget || query.UnifiedAlertingEnabled {
 			// get quota used.
 			rawSQL := fmt.Sprintf("SELECT COUNT(*) AS count FROM %s",
 				dialect.Quote(query.Target))

pkg/services/store/http.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/grafana/grafana/pkg/api/response"
 	"github.com/grafana/grafana/pkg/models"
+	"github.com/grafana/grafana/pkg/services/quota"
 	"github.com/grafana/grafana/pkg/util"
 	"github.com/grafana/grafana/pkg/web"
 )
@@ -26,12 +27,14 @@ type HTTPStorageService interface {
 }
 
 type httpStorage struct {
-	store StorageService
+	store        StorageService
+	quotaService quota.Service
 }
 
-func ProvideHTTPService(store StorageService) HTTPStorageService {
+func ProvideHTTPService(store StorageService, quotaService quota.Service) HTTPStorageService {
 	return &httpStorage{
-		store: store,
+		store:        store,
+		quotaService: quotaService,
 	}
 }
 
@@ -58,6 +61,16 @@ func UploadErrorToStatusCode(err error) int {
 }
 
 func (s *httpStorage) Upload(c *models.ReqContext) response.Response {
+	// assumes we are only uploading to the SQL database - TODO: refactor once we introduce object stores
+	quotaReached, err := s.quotaService.CheckQuotaReached(c.Req.Context(), "file", nil)
+	if err != nil {
+		return response.Error(500, "Internal server error", err)
+	}
+
+	if quotaReached {
+		return response.Error(400, "File quota reached", errors.New("file quota reached"))
+	}
+
 	type rspInfo struct {
 		Message string `json:"message,omitempty"`
 		Path    string `json:"path,omitempty"`

pkg/setting/setting_quota.go

@@ -24,6 +24,7 @@ type GlobalQuota struct {
 	ApiKey     int64 `target:"api_key"`
 	Session    int64 `target:"-"`
 	AlertRule  int64 `target:"alert_rule"`
+	File       int64 `target:"file"`
 }
 
 func (q *OrgQuota) ToMap() map[string]int64 {
@@ -94,6 +95,7 @@ func (cfg *Cfg) readQuotaSettings() {
 		Dashboard:  quota.Key("global_dashboard").MustInt64(-1),
 		ApiKey:     quota.Key("global_api_key").MustInt64(-1),
 		Session:    quota.Key("global_session").MustInt64(-1),
+		File:       quota.Key("global_file").MustInt64(-1),
 		AlertRule:  alertGlobalQuota,
 	}