K8s Apiserver max-requests-inflight 流控实现

func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler {
	handler := genericapifilters.WithAuthorization(apiHandler, c.Authorization.Authorizer, c.Serializer)
	handler = genericfilters.WithMaxInFlightLimit(handler, c.MaxRequestsInFlight, c.MaxMutatingRequestsInFlight, c.LongRunningFunc)
	handler = genericapifilters.WithImpersonation(handler, c.Authorization.Authorizer, c.Serializer)
	handler = genericapifilters.WithAudit(handler, c.AuditBackend, c.AuditPolicyChecker, c.LongRunningFunc)
	failedHandler := genericapifilters.Unauthorized(c.Serializer, c.Authentication.SupportsBasicAuth)
	failedHandler = genericapifilters.WithFailedAuthenticationAudit(failedHandler, c.AuditBackend, c.AuditPolicyChecker)
	handler = genericapifilters.WithAuthentication(handler, c.Authentication.Authenticator, failedHandler, c.Authentication.APIAudiences)
	handler = genericfilters.WithCORS(handler, c.CorsAllowedOriginList, nil, nil, nil, "true")
	handler = genericfilters.WithTimeoutForNonLongRunningRequests(handler, c.LongRunningFunc, c.RequestTimeout)
	handler = genericfilters.WithWaitGroup(handler, c.LongRunningFunc, c.HandlerChainWaitGroup)
	handler = genericapifilters.WithRequestInfo(handler, c.RequestInfoResolver)
	handler = genericapifilters.WithCacheControl(handler)
	handler = genericfilters.WithPanicRecovery(handler)
	return handler
}

func WithMaxInFlightLimit(handler http.Handler) http.Handler {
    
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        // before 处理逻辑
        // ... ...

        handler.ServeHTTP(w, r)
        
        // after 处理逻辑
        // ... ...
    }
}

// WithMaxInFlightLimit limits the number of in-flight requests to buffer size of the passed in channel.
func WithMaxInFlightLimit(
	handler http.Handler,
	nonMutatingLimit int,
	mutatingLimit int,
	longRunningRequestCheck apirequest.LongRunningRequestCheck,
) http.Handler {
	startOnce.Do(startRecordingUsage)
	if nonMutatingLimit == 0 && mutatingLimit == 0 {
		return handler
    }
    
    // 声明的全局的 channel，用来作为请求的阻塞队列，初始化大小
	var nonMutatingChan chan bool
	var mutatingChan chan bool
	if nonMutatingLimit != 0 {
		nonMutatingChan = make(chan bool, nonMutatingLimit)
	}
	if mutatingLimit != 0 {
		mutatingChan = make(chan bool, mutatingLimit)
	}

    // 中间件的返回参数，也是一个 handler，封装原来的 handler
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		ctx := r.Context()
		requestInfo, ok := apirequest.RequestInfoFrom(ctx)
		if !ok {
			handleError(w, r, fmt.Errorf("no RequestInfo found in context, handler chain must be wrong"))
			return
		}

		// Skip tracking long running events.
		if longRunningRequestCheck != nil && longRunningRequestCheck(r, requestInfo) {
			handler.ServeHTTP(w, r)
			return
		}

		var c chan bool
		isMutatingRequest := !nonMutatingRequestVerbs.Has(requestInfo.Verb)
		if isMutatingRequest {
			c = mutatingChan
		} else {
			c = nonMutatingChan
		}

		if c == nil { // 这里是对应的 verb 没有设置 limit（比如是 mutate 请求，但是mutatingLimit参数置为0），直接转发到原来的 http handler 处理，不做限制
			handler.ServeHTTP(w, r)
		} else {

			select {
                //尝试向阻塞队列里写数据，能写进去，表示正在处理的请求没有把 channel 打满
			case c <- true:
				var mutatingLen, readOnlyLen int
				if isMutatingRequest {
					mutatingLen = len(mutatingChan)
				} else {
					readOnlyLen = len(nonMutatingChan)
				}

				defer func() {
                    // handler 处理完了之后，从阻塞队列 channel 中读出一个元素，表示正在处理的请求少了一个
					<-c
					if isMutatingRequest {
						watermark.recordMutating(mutatingLen)
					} else {
						watermark.recordReadOnly(readOnlyLen)
					}

                }()
                // 能写进去说明当前未达到限制，可以直接处理
				handler.ServeHTTP(w, r)

			default:
                // 省去了跟监控相关的代码。

                // 下面的 tooManyRequest 就是 channel 写不进去的处理逻辑，也就是当前所有处理请求数量已经达到限制了， 这个限制就是 channel 的长度。
				tooManyRequests(r, w)
			}
		}
	})
}

// 处理请求太多了怎么办？返回错误码、header、body，直接返回
func tooManyRequests(req *http.Request, w http.ResponseWriter) {
	// Return a 429 status indicating "Too Many Requests"
	w.Header().Set("Retry-After", acquireReteyAfterTime())
	http.Error(w, "Too many requests, please try again later.", http.StatusTooManyRequests)
}

// 下面是 http.Error() 的实现
// Error replies to the request with the specified error message and HTTP code.
// It does not otherwise end the request; the caller should ensure no further
// writes are done to w.
// The error message should be plain text.
func Error(w ResponseWriter, error string, code int) {
	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
	w.Header().Set("X-Content-Type-Options", "nosniff")
	w.WriteHeader(code)
	fmt.Fprintln(w, error)
}