QoS

api-server.go

@@ -67,6 +67,8 @@ func configureServerHandler(c *ServerConfig) http.Handler {
 		api.SetIgnoreResourcesHandler,
 		// Add new handlers here.
 
+		api.SetQosHandler,
+
 		api.SetLogHandler,
 
 		api.NewAccessLogHandler,

api/generic-handlers.go

@@ -19,16 +19,15 @@ package api
 import (
 	"context"
 	"fmt"
-	"net/http"
-	"strings"
-
 	"github.com/gorilla/mux"
 	. "github.com/journeymidnight/yig/error"
 	"github.com/journeymidnight/yig/helper"
 	"github.com/journeymidnight/yig/log"
 	"github.com/journeymidnight/yig/meta"
 	"github.com/journeymidnight/yig/meta/types"
 	"github.com/journeymidnight/yig/signature"
+	"net/http"
+	"strings"
 )
 
 // HandlerFunc - useful to chain different middleware http.Handler
@@ -273,6 +272,38 @@ func InReservedOrigins(origin string) bool {
 	return false
 }
 
+type QosHandler struct {
+	handler http.Handler
+	meta    *meta.Meta
+}
+
+func (h *QosHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	ctx := getRequestContext(r)
+	if len(ctx.BucketName) == 0 {
+		h.handler.ServeHTTP(w, r)
+		return
+	}
+	var allow bool
+	if r.Method == "GET" || r.Method == "HEAD" { // read operations
+		allow = h.meta.QosMeta.AllowReadQuery(ctx.BucketName)
+	} else { // write operations
+		allow = h.meta.QosMeta.AllowWriteQuery(ctx.BucketName)
+	}
+	if !allow {
+		WriteErrorResponse(w, r, ErrRequestLimitExceeded)
+		return
+	}
+	h.handler.ServeHTTP(w, r)
+}
+
+func SetQosHandler(h http.Handler, meta *meta.Meta) http.Handler {
+	qos := QosHandler{
+		handler: h,
+		meta:    meta,
+	}
+	return &qos
+}
+
 //// helpers
 
 func GetBucketAndObjectInfoFromRequest(r *http.Request) (bucketName string, objectName string, isBucketDomain bool) {
@@ -300,7 +331,7 @@ func getRequestContext(r *http.Request) RequestContext {
 		return ctx
 	}
 	return RequestContext{
-		Logger: r.Context().Value(ContextLoggerKey).(log.Logger),
+		Logger:    r.Context().Value(ContextLoggerKey).(log.Logger),
 		RequestID: r.Context().Value(RequestIdKey).(string),
 	}
 }

error/api-errors.go

@@ -177,6 +177,7 @@ const (
 	ErrInvalidRestoreInfo
 	ErrCreateRestoreObject
 	ErrInvalidGlacierObject
+	ErrRequestLimitExceeded
 )
 
 // error code to APIError structure, these fields carry respective
@@ -828,6 +829,11 @@ var ErrorCodeResponse = map[ApiErrorCode]ApiErrorStruct{
 		Description:    "Create object thaw operation failed",
 		HttpStatusCode: http.StatusInternalServerError,
 	},
+	ErrRequestLimitExceeded: {
+		AwsErrorCode:   "ErrRequestLimitExceeded",
+		Description:    "Request limit exceeded",
+		HttpStatusCode: http.StatusTooManyRequests,
+	},
 }
 
 func (e ApiErrorCode) AwsErrorCode() string {

go.mod

@@ -30,9 +30,11 @@ require (
 	github.com/BurntSushi/toml v0.3.1
 	github.com/DATA-DOG/go-sqlmock v1.3.3
 	github.com/cep21/circuit v0.0.0-20181030180945-e893c027dc21
-	github.com/confluentinc/confluent-kafka-go v1.0.0 //indirect
+	github.com/confluentinc/confluent-kafka-go v1.0.0 //ct
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/dustin/go-humanize v1.0.0
+	github.com/go-redis/redis/v7 v7.0.0-beta.4
+	github.com/go-redis/redis_rate/v8 v8.0.0
 	github.com/go-sql-driver/mysql v1.4.1
 	github.com/golang/snappy v0.0.1
 	github.com/gomodule/redigo v1.7.0
@@ -42,7 +44,7 @@ require (
 	github.com/journeymidnight/radoshttpd v0.0.0-20190617133011-609666b51136
 	github.com/minio/highwayhash v1.0.0
 	github.com/prometheus/client_golang v0.9.3-0.20190127221311-3c4408c8b829
-	github.com/stretchr/testify v1.3.0
+	github.com/stretchr/testify v1.4.0
 	github.com/ugorji/go v1.1.4
 	github.com/xxtea/xxtea-go v0.0.0-20170828040851-35c4b17eecf6
 	golang.org/x/crypto v0.0.0-20190325154230-a5d413f7728c // indirect

go.sum

@@ -17,8 +17,13 @@ github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumC
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
+github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
+github.com/go-redis/redis/v7 v7.0.0-beta.4 h1:p6z7Pde69EGRWvlC++y8aFcaWegyrKHzOBGo0zUACTQ=
+github.com/go-redis/redis/v7 v7.0.0-beta.4/go.mod h1:xhhSbUMTsleRPur+Vgx9sUHtyN33bdjxY+9/0n9Ig8s=
+github.com/go-redis/redis_rate/v8 v8.0.0 h1:V6ZQWFLDECyeUJ30LRzfTM3fx5GdERYC0cnTSY5z0KE=
+github.com/go-redis/redis_rate/v8 v8.0.0/go.mod h1:4ZBS7uoZS1Y/ZBMFMlMNBt1W0rU7vwfnpZku3FpjlfM=
 github.com/go-sql-driver/mysql v1.4.1 h1:g24URVg0OFbNUTx9qqY1IRZ9D9z3iPyi5zKhQZpNwpA=
 github.com/go-sql-driver/mysql v1.4.1/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
@@ -45,6 +50,7 @@ github.com/gorilla/context v1.1.1 h1:AWwleXJkX/nhcU9bZSnZoi3h/qGYqQAGhq6zZe/aQW8
 github.com/gorilla/context v1.1.1/go.mod h1:kBGZzfjB9CEq2AlWe17Uuf7NDRt0dE0s8S51q0aT7Yg=
 github.com/gorilla/mux v1.6.2 h1:Pgr17XVTNXAk3q/r4CpKzC5xBM/qW1uVLV+IhRZpIIk=
 github.com/gorilla/mux v1.6.2/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs=
+github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af h1:pmfjZENx5imkbgOkpRUYLnmbU7UEFbjtDA2hxJ1ichM=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
 github.com/journeymidnight/aws-sdk-go v1.18.1 h1:/dv12U8x+EHVEmW66rXtI00Do+ZMREGitokPGMpUalc=
@@ -60,6 +66,9 @@ github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5
 github.com/minio/highwayhash v1.0.0 h1:iMSDhgUILCr0TNm8LWlSjF8N0ZIj2qbO8WHp6Q/J2BA=
 github.com/minio/highwayhash v1.0.0/go.mod h1:xQboMTeM9nY9v/LlAOxFctujiv5+Aq2hR5dxBpaMbdc=
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
+github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/ginkgo v1.8.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/gomega v1.5.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -83,11 +92,15 @@ github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoH
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/ugorji/go v1.1.4 h1:j4s+tAvLfL3bZyefP2SEWmhBzmuIlH/eqNuPdFPgngw=
 github.com/ugorji/go v1.1.4/go.mod h1:uQMGLiO92mf5W77hV/PUCpI3pbzQx3CRekS0kk+RGrc=
 github.com/xxtea/xxtea-go v0.0.0-20170828040851-35c4b17eecf6 h1:S+0oS/OPAe0kdSpQ7GAnCmpcDL7Jh2iJMjZTV6mYbPo=
 github.com/xxtea/xxtea-go v0.0.0-20170828040851-35c4b17eecf6/go.mod h1:2uvuCBt0VXxijrX5ieiAeeNT2+2MIsrs1DI9iXz7OOQ=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

integrate/yig.sql

@@ -273,6 +273,15 @@ CREATE TABLE `users` (
   `userid` varchar(255) DEFAULT NULL,
   `bucketname` varchar(255) DEFAULT NULL
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
+
+DROP TABLE IF EXISTS `qos`;
+CREATE TABLE `qos` (
+    `userid` varchar(255) PRIMARY KEY NOT NULL DEFAULT '',
+    `read_qps` int NOT NULL DEFAULT 0 COMMENT 'read query per second',
+    `write_qps` int NOT NULL DEFAULT 0 COMMENT 'write query per second',
+    `bandwidth` int NOT NULL DEFAULT 0 COMMENT 'in KiB per second'
+);
+
 /*!40101 SET character_set_client = @saved_cs_client */;
 /*!40103 SET TIME_ZONE=@OLD_TIME_ZONE */;
 

meta/client/client.go

@@ -51,6 +51,7 @@ type Client interface {
 	ScanLifeCycle(limit int, marker string) (result ScanLifeCycleResult, err error)
 	//user
 	GetUserBuckets(userId string) (buckets []string, err error)
+	GetAllUserBuckets() (bucketUser map[string]string, err error)
 	AddBucketForUser(bucketName, userId string) (err error)
 	RemoveBucketForUser(bucketName string, userId string) (err error)
 	//gc
@@ -64,4 +65,6 @@ type Client interface {
 	GetFreezerStatus(bucketName, objectName, version string) (freezer *Freezer, err error)
 	UploadFreezerDate(bucketName, objectName string, lifetime int) (err error)
 	DeleteFreezer(bucketName, objectName string, tx DB) (err error)
+	//qos
+	GetAllUserQos() (userQos map[string]UserQos, err error)
 }

meta/client/tidbclient/qos.go

@@ -0,0 +1,25 @@
+package tidbclient
+
+import (
+	. "github.com/journeymidnight/yig/meta/types"
+)
+
+func (t *TidbClient) GetAllUserQos() (userQos map[string]UserQos, err error) {
+	userQos = make(map[string]UserQos)
+	rows, err := t.Client.Query( `select userid, read_qps, write_qps, bandwidth 
+		from qos`)
+	if err != nil {
+		return
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var qos UserQos
+		err = rows.Scan(&qos.UserID, &qos.ReadQps, &qos.WriteQps, &qos.Bandwidth)
+		if err != nil {
+			return
+		}
+		userQos[qos.UserID] = qos
+	}
+	return
+}

meta/client/tidbclient/user.go

@@ -26,6 +26,26 @@ func (t *TidbClient) GetUserBuckets(userId string) (buckets []string, err error)
 	return
 }
 
+func (t *TidbClient) GetAllUserBuckets() (bucketUser map[string]string, err error) {
+	// bucket name -> user id
+	bucketUser = make(map[string]string)
+	rows, err := t.Client.Query("select userid, bucketname from users")
+	if err != nil {
+		return
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var userID, bucketName string
+		err = rows.Scan(&userID, &bucketName)
+		if err != nil {
+			return
+		}
+		bucketUser[bucketName] = userID
+	}
+	return
+}
+
 func (t *TidbClient) AddBucketForUser(bucketName, userId string) (err error) {
 	sql := "insert into users(userid,bucketname) values(?,?)"
 	_, err = t.Client.Exec(sql, userId, bucketName)

meta/meta.go

@@ -11,18 +11,20 @@ const (
 )
 
 type Meta struct {
-	Client client.Client
-	Cache  MetaCache
+	Client  client.Client
+	Cache   MetaCache
+	QosMeta *QosMeta
 }
 
 func New(myCacheType CacheType) *Meta {
 	meta := Meta{
-		Cache:  newMetaCache(myCacheType),
+		Cache: newMetaCache(myCacheType),
 	}
 	if helper.CONFIG.MetaStore == "tidb" {
 		meta.Client = tidbclient.NewTidbClient()
 	} else {
 		panic("unsupport metastore")
 	}
+	meta.QosMeta = NewQosMeta(meta.Client)
 	return &meta
-}
+}