Add rate limiter instead of connection limit for now

This commit is contained in:
Harshavardhana 2015-06-06 17:01:39 -07:00
parent 996c1a5c28
commit 371651bde6
7 changed files with 86 additions and 74 deletions

14
main.go
View File

@ -46,9 +46,9 @@ var flags = []cli.Flag{
}, },
*/ */
cli.IntFlag{ cli.IntFlag{
Name: "conn-limit", Name: "ratelimit",
Value: 16, Value: 16,
Usage: "Set per IP connection limit quota for server: [DEFAULT: 16]", Usage: "Limit for total concurrent requests: [DEFAULT: 16]",
}, },
cli.StringFlag{ cli.StringFlag{
Name: "cert", Name: "cert",
@ -84,11 +84,11 @@ func getAPIServerConfig(c *cli.Context) httpserver.Config {
} }
tls := (certFile != "" && keyFile != "") tls := (certFile != "" && keyFile != "")
return httpserver.Config{ return httpserver.Config{
Address: c.GlobalString("address"), Address: c.GlobalString("address"),
TLS: tls, TLS: tls,
CertFile: certFile, CertFile: certFile,
KeyFile: keyFile, KeyFile: keyFile,
ConnectionLimit: c.GlobalInt("conn-limit"), RateLimit: c.GlobalInt("ratelimit"),
} }
} }

View File

@ -31,8 +31,8 @@ type minioAPI struct {
// Config api configurable parameters // Config api configurable parameters
type Config struct { type Config struct {
ConnectionLimit int RateLimit int
driver drivers.Driver driver drivers.Driver
} }
// GetDriver - get a an existing set driver // GetDriver - get a an existing set driver
@ -69,11 +69,12 @@ func HTTPHandler(config Config) http.Handler {
handler = timeValidityHandler(handler) handler = timeValidityHandler(handler)
handler = ignoreResourcesHandler(handler) handler = ignoreResourcesHandler(handler)
handler = validateAuthHeaderHandler(handler) handler = validateAuthHeaderHandler(handler)
// h = quota.BandwidthCap(h, 25*1024*1024, time.Duration(30*time.Minute)) // handler = quota.BandwidthCap(h, 25*1024*1024, time.Duration(30*time.Minute))
// h = quota.BandwidthCap(h, 100*1024*1024, time.Duration(24*time.Hour)) // handler = quota.BandwidthCap(h, 100*1024*1024, time.Duration(24*time.Hour))
// h = quota.RequestLimit(h, 100, time.Duration(30*time.Minute)) // handler = quota.RequestLimit(h, 100, time.Duration(30*time.Minute))
// h = quota.RequestLimit(h, 1000, time.Duration(24*time.Hour)) // handler = quota.RequestLimit(h, 1000, time.Duration(24*time.Hour))
handler = quota.ConnectionLimit(handler, config.ConnectionLimit) // handler = quota.ConnectionLimit(handler, config.ConnectionLimit)
handler = quota.RateLimit(handler, config.RateLimit)
handler = logging.LogHandler(handler) handler = logging.LogHandler(handler)
return handler return handler
} }

View File

@ -124,7 +124,7 @@ func setDummyAuthHeader(req *http.Request) {
} }
func setConfig(driver drivers.Driver) Config { func setConfig(driver drivers.Driver) Config {
conf := Config{ConnectionLimit: 16} conf := Config{RateLimit: 16}
conf.SetDriver(driver) conf.SetDriver(driver)
return conf return conf
} }

View File

@ -16,74 +16,35 @@
package quota package quota
import ( import "net/http"
"net"
"net/http"
"sync"
"github.com/minio/minio/pkg/utils/log"
)
// requestLimitHandler // requestLimitHandler
type connLimit struct { type connLimit struct {
sync.RWMutex handler http.Handler
handler http.Handler connectionQueue chan bool
connections map[uint32]int
limit int
} }
func (c *connLimit) IsLimitExceeded(ip uint32) bool { func (c *connLimit) Add() {
if c.connections[ip] >= c.limit { c.connectionQueue <- true
return true
}
return false
}
func (c *connLimit) GetUsed(ip uint32) int {
return c.connections[ip]
}
func (c *connLimit) Add(ip uint32) {
c.Lock()
defer c.Unlock()
count := c.connections[ip]
count = count + 1
c.connections[ip] = count
return return
} }
func (c *connLimit) Remove(ip uint32) { func (c *connLimit) Remove() {
c.Lock() <-c.connectionQueue
defer c.Unlock() return
count, _ := c.connections[ip]
count = count - 1
if count <= 0 {
delete(c.connections, ip)
return
}
c.connections[ip] = count
} }
// ServeHTTP is an http.Handler ServeHTTP method // ServeHTTP is an http.Handler ServeHTTP method
func (c *connLimit) ServeHTTP(w http.ResponseWriter, req *http.Request) { func (c *connLimit) ServeHTTP(w http.ResponseWriter, req *http.Request) {
host, _, _ := net.SplitHostPort(req.RemoteAddr) c.Add()
longIP := longIP{net.ParseIP(host)}.IptoUint32()
if c.IsLimitExceeded(longIP) {
hosts, _ := net.LookupAddr(uint32ToIP(longIP).String())
log.Debug.Printf("Connection limit reached - Host: %s, Total Connections: %d\n", hosts, c.GetUsed(longIP))
writeErrorResponse(w, req, ConnectionLimitExceeded, req.URL.Path)
return
}
c.Add(longIP)
defer c.Remove(longIP)
c.handler.ServeHTTP(w, req) c.handler.ServeHTTP(w, req)
c.Remove()
} }
// ConnectionLimit limits the number of concurrent connections // ConnectionLimit limits the number of concurrent connections
func ConnectionLimit(h http.Handler, limit int) http.Handler { func ConnectionLimit(h http.Handler, limit int) http.Handler {
return &connLimit{ return &connLimit{
handler: h, handler: h,
connections: make(map[uint32]int), connectionQueue: make(chan bool, limit),
limit: limit,
} }
} }

View File

@ -0,0 +1,50 @@
/*
* Minimalist Object Storage, (C) 2015 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package quota
import "net/http"
// rateLimit
type rateLimit struct {
handler http.Handler
rateQueue chan bool
}
func (c *rateLimit) Add() {
c.rateQueue <- true // fill in the queue
return
}
func (c *rateLimit) Remove() {
<-c.rateQueue // invalidate the queue, after the request is served
return
}
// ServeHTTP is an http.Handler ServeHTTP method
func (c *rateLimit) ServeHTTP(w http.ResponseWriter, req *http.Request) {
c.Add() // add
c.handler.ServeHTTP(w, req) // serve
c.Remove() // remove
}
// RateLimit limits the number of concurrent http requests
func RateLimit(handle http.Handler, limit int) http.Handler {
return &rateLimit{
handler: handle,
rateQueue: make(chan bool, limit),
}
}

View File

@ -25,11 +25,11 @@ import (
// Config - http server config // Config - http server config
type Config struct { type Config struct {
Address string Address string
TLS bool TLS bool
CertFile string CertFile string
KeyFile string KeyFile string
ConnectionLimit int RateLimit int
} }
// Server - http server related // Server - http server related

View File

@ -43,7 +43,7 @@ type MemoryFactory struct {
func (f MemoryFactory) GetStartServerFunc() StartServerFunc { func (f MemoryFactory) GetStartServerFunc() StartServerFunc {
return func() (chan<- string, <-chan error) { return func() (chan<- string, <-chan error) {
_, _, driver := memory.Start(f.MaxMemory, f.Expiration) _, _, driver := memory.Start(f.MaxMemory, f.Expiration)
conf := api.Config{ConnectionLimit: f.ConnectionLimit} conf := api.Config{RateLimit: f.RateLimit}
conf.SetDriver(driver) conf.SetDriver(driver)
ctrl, status, _ := httpserver.Start(api.HTTPHandler(conf), f.Config) ctrl, status, _ := httpserver.Start(api.HTTPHandler(conf), f.Config)
return ctrl, status return ctrl, status
@ -60,7 +60,7 @@ type FilesystemFactory struct {
func (f FilesystemFactory) GetStartServerFunc() StartServerFunc { func (f FilesystemFactory) GetStartServerFunc() StartServerFunc {
return func() (chan<- string, <-chan error) { return func() (chan<- string, <-chan error) {
_, _, driver := fs.Start(f.Path) _, _, driver := fs.Start(f.Path)
conf := api.Config{ConnectionLimit: f.ConnectionLimit} conf := api.Config{RateLimit: f.RateLimit}
conf.SetDriver(driver) conf.SetDriver(driver)
ctrl, status, _ := httpserver.Start(api.HTTPHandler(conf), f.Config) ctrl, status, _ := httpserver.Start(api.HTTPHandler(conf), f.Config)
return ctrl, status return ctrl, status
@ -90,7 +90,7 @@ type DonutFactory struct {
func (f DonutFactory) GetStartServerFunc() StartServerFunc { func (f DonutFactory) GetStartServerFunc() StartServerFunc {
return func() (chan<- string, <-chan error) { return func() (chan<- string, <-chan error) {
_, _, driver := donut.Start(f.Paths) _, _, driver := donut.Start(f.Paths)
conf := api.Config{ConnectionLimit: f.ConnectionLimit} conf := api.Config{RateLimit: f.RateLimit}
conf.SetDriver(driver) conf.SetDriver(driver)
ctrl, status, _ := httpserver.Start(api.HTTPHandler(conf), f.Config) ctrl, status, _ := httpserver.Start(api.HTTPHandler(conf), f.Config)
return ctrl, status return ctrl, status