oxy 学习笔记

oxy

作用

限流 负载 代理

Oxy is a Go library with HTTP handlers that enhance HTTP standard library:

• Buffer retries and buffers requests and responses
• Stream passes-through requests, supports chunked encoding with configurable flush interval
• Forward forwards requests to remote location and rewrites headers
• Roundrobin is a round-robin load balancer
• Circuit Breaker Hystrix-style circuit breaker
• Connlimit Simultaneous connections limiter
• Ratelimit Rate limiter (based on tokenbucket algo)
• Trace Structured request and response logger

Forward

核心代码

    // Modify the request to handle the target URL
    func (f *httpForwarder) modifyRequest(outReq *http.Request, target *url.URL) {
        outReq.URL = utils.CopyURL(outReq.URL)
        outReq.URL.Scheme = target.Scheme
        outReq.URL.Host = target.Host

        u := f.getUrlFromRequest(outReq)

        outReq.URL.Path = u.Path
        outReq.URL.RawPath = u.RawPath
        outReq.URL.RawQuery = u.RawQuery
        outReq.RequestURI = "" // Outgoing request should not have RequestURI

        outReq.Proto = "HTTP/1.1"
        outReq.ProtoMajor = 1
        outReq.ProtoMinor = 1

        if f.rewriter != nil {
            f.rewriter.Rewrite(outReq)
        }

        // Do not pass client Host header unless optsetter PassHostHeader is set.
        if !f.passHost {
            outReq.Host = target.Host
        }
    }

	revproxy := httputil.ReverseProxy{
		Director: func(req *http.Request) {
			f.modifyRequest(req, inReq.URL)
		},
		Transport:      f.roundTripper,
		FlushInterval:  f.flushInterval,
		ModifyResponse: f.modifyResponse,
		BufferPool:     f.bufferPool,
    }
    
    revproxy.ServeHTTP(w, outReq)

ratelimit

核心代码

func (tl *TokenLimiter) ServeHTTP(w http.ResponseWriter, req *http.Request) {
    // 提取ip,host 信息 
	source, amount, err := tl.extract.Extract(req)
	if err != nil {
		tl.errHandler.ServeHTTP(w, req, err)
		return
	}

    // 消费rate
	if err := tl.consumeRates(req, source, amount); err != nil {
		tl.log.Warnf("limiting request %v %v, limit: %v", req.Method, req.URL, err)
		tl.errHandler.ServeHTTP(w, req, err)
		return
	}

	tl.next.ServeHTTP(w, req)
}

func (tl *TokenLimiter) consumeRates(req *http.Request, source string, amount int64) error {
	tl.mutex.Lock()
	defer tl.mutex.Unlock()

	effectiveRates := tl.resolveRates(req)
	bucketSetI, exists := tl.bucketSets.Get(source)
	var bucketSet *TokenBucketSet

	if exists {
		bucketSet = bucketSetI.(*TokenBucketSet)
		bucketSet.Update(effectiveRates)
	} else {
		bucketSet = NewTokenBucketSet(effectiveRates, tl.clock)
		// We set ttl as 10 times rate period. E.g. if rate is 100 requests/second per client ip
		// the counters for this ip will expire after 10 seconds of inactivity
		tl.bucketSets.Set(source, bucketSet, int(bucketSet.maxPeriod/time.Second)*10+1)
	}
	delay, err := bucketSet.Consume(amount)
	if err != nil {
		return err
	}
	if delay > 0 {
		return &MaxRateError{Delay: delay}
	}
	return nil
}