kubernetes
/
apiserver
mirror of https://github.com/kubernetes/apiserver.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Rename width to workEstimate in P&F code 

Kubernetes-commit: 73211256e8f15cf84ee69d6fe8258c3a912e0f94
This commit is contained in:
wojtekt 2021-07-13 15:10:58 +02:00 committed by Kubernetes Publisher
parent 4feaaf7955
commit b4c306e1e8
13 changed files with 69 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
pkg/server/config.go
View File

@ -222,9 +222,6 @@ type Config struct {
// If not specify any in flags, then genericapiserver will only enable defaultAPIResourceConfig. // If not specify any in flags, then genericapiserver will only enable defaultAPIResourceConfig.
MergedResourceConfig *serverstore.ResourceConfig MergedResourceConfig *serverstore.ResourceConfig
// RequestWidthEstimator is used to estimate the "width" of the incoming request(s).
RequestWidthEstimator flowcontrolrequest.WidthEstimatorFunc
// lifecycleSignals provides access to the various signals // lifecycleSignals provides access to the various signals
// that happen during lifecycle of the apiserver. // that happen during lifecycle of the apiserver.
// it's intentionally marked private as it should never be overridden. // it's intentionally marked private as it should never be overridden.
@ -352,9 +349,8 @@ func NewConfig(codecs serializer.CodecFactory) *Config {
// Default to treating watch as a long-running operation // Default to treating watch as a long-running operation
// Generic API servers have no inherent long-running subresources // Generic API servers have no inherent long-running subresources
LongRunningFunc: genericfilters.BasicLongRunningRequestCheck(sets.NewString("watch"), sets.NewString()), LongRunningFunc: genericfilters.BasicLongRunningRequestCheck(sets.NewString("watch"), sets.NewString()),
RequestWidthEstimator: flowcontrolrequest.DefaultWidthEstimator, lifecycleSignals: newLifecycleSignals(),
lifecycleSignals: newLifecycleSignals(),
APIServerID: id, APIServerID: id,
StorageVersionManager: storageversion.NewDefaultManager(), StorageVersionManager: storageversion.NewDefaultManager(),
@ -747,7 +743,7 @@ func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler {
if c.FlowControl != nil { if c.FlowControl != nil {
handler = filterlatency.TrackCompleted(handler) handler = filterlatency.TrackCompleted(handler)
handler = genericfilters.WithPriorityAndFairness(handler, c.LongRunningFunc, c.FlowControl, c.RequestWidthEstimator) handler = genericfilters.WithPriorityAndFairness(handler, c.LongRunningFunc, c.FlowControl, flowcontrolrequest.DefaultWorkEstimator)
handler = filterlatency.TrackStarted(handler, "priorityandfairness") handler = filterlatency.TrackStarted(handler, "priorityandfairness")
} else { } else {
handler = genericfilters.WithMaxInFlightLimit(handler, c.MaxRequestsInFlight, c.MaxMutatingRequestsInFlight, c.LongRunningFunc) handler = genericfilters.WithMaxInFlightLimit(handler, c.MaxRequestsInFlight, c.MaxMutatingRequestsInFlight, c.LongRunningFunc)

13
pkg/server/filters/priority-and-fairness.go
View File

@ -61,7 +61,7 @@ func WithPriorityAndFairness(
handler http.Handler, handler http.Handler,
longRunningRequestCheck apirequest.LongRunningRequestCheck, longRunningRequestCheck apirequest.LongRunningRequestCheck,
fcIfc utilflowcontrol.Interface, fcIfc utilflowcontrol.Interface,
widthEstimator flowcontrolrequest.WidthEstimatorFunc, workEstimator flowcontrolrequest.WorkEstimatorFunc,
) http.Handler { ) http.Handler {
if fcIfc == nil { if fcIfc == nil {
klog.Warningf("priority and fairness support not found, skipping") klog.Warningf("priority and fairness support not found, skipping")
@ -122,11 +122,14 @@ func WithPriorityAndFairness(
} }
} }
// find the estimated "width" of the request // find the estimated amount of work of the request
// TODO: Maybe just make it costEstimator and let it return additionalLatency too for the watch?
// TODO: Estimate cost should also take fcIfc.GetWatchCount(requestInfo) as a parameter. // TODO: Estimate cost should also take fcIfc.GetWatchCount(requestInfo) as a parameter.
width := widthEstimator.EstimateWidth(r) workEstimate := workEstimator.EstimateWork(r)
digest := utilflowcontrol.RequestDigest{RequestInfo: requestInfo, User: user, Width: width} digest := utilflowcontrol.RequestDigest{
RequestInfo: requestInfo,
User: user,
WorkEstimate: workEstimate,
}
if isWatchRequest { if isWatchRequest {
// This channel blocks calling handler.ServeHTTP() until closed, and is closed inside execute(). // This channel blocks calling handler.ServeHTTP() until closed, and is closed inside execute().

12
pkg/server/filters/priority-and-fairness_test.go
View File

@ -70,7 +70,9 @@ const (
decisionSkipFilter decisionSkipFilter
) )
var defaultRequestWidthEstimator = func(*http.Request) fcrequest.Width { return fcrequest.Width{Seats: 1} } var defaultRequestWorkEstimator = func(*http.Request) fcrequest.WorkEstimate {
return fcrequest.WorkEstimate{Seats: 1}
}
type fakeApfFilter struct { type fakeApfFilter struct {
mockDecision mockDecision mockDecision mockDecision
@ -165,7 +167,7 @@ func newApfHandlerWithFilter(t *testing.T, flowControlFilter utilflowcontrol.Int
apfHandler := WithPriorityAndFairness(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { apfHandler := WithPriorityAndFairness(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
onExecute() onExecute()
}), longRunningRequestCheck, flowControlFilter, defaultRequestWidthEstimator) }), longRunningRequestCheck, flowControlFilter, defaultRequestWorkEstimator)
handler := apifilters.WithRequestInfo(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { handler := apifilters.WithRequestInfo(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
r = r.WithContext(apirequest.WithUser(r.Context(), &user.DefaultInfo{ r = r.WithContext(apirequest.WithUser(r.Context(), &user.DefaultInfo{
@ -649,7 +651,7 @@ func TestApfWithRequestDigest(t *testing.T) {
reqDigestExpected := &utilflowcontrol.RequestDigest{ reqDigestExpected := &utilflowcontrol.RequestDigest{
RequestInfo: &apirequest.RequestInfo{Verb: "get"}, RequestInfo: &apirequest.RequestInfo{Verb: "get"},
User: &user.DefaultInfo{Name: "foo"}, User: &user.DefaultInfo{Name: "foo"},
Width: fcrequest.Width{ WorkEstimate: fcrequest.WorkEstimate{
Seats: 5, Seats: 5,
}, },
} }
@ -657,7 +659,7 @@ func TestApfWithRequestDigest(t *testing.T) {
handler := WithPriorityAndFairness(http.HandlerFunc(func(_ http.ResponseWriter, req *http.Request) {}), handler := WithPriorityAndFairness(http.HandlerFunc(func(_ http.ResponseWriter, req *http.Request) {}),
longRunningFunc, longRunningFunc,
fakeFilter, fakeFilter,
func(_ *http.Request) fcrequest.Width { return reqDigestExpected.Width }, func(_ *http.Request) fcrequest.WorkEstimate { return reqDigestExpected.WorkEstimate },
) )
w := httptest.NewRecorder() w := httptest.NewRecorder()
@ -1171,7 +1173,7 @@ func newHandlerChain(t *testing.T, handler http.Handler, filter utilflowcontrol.
requestInfoFactory := &apirequest.RequestInfoFactory{APIPrefixes: sets.NewString("apis", "api"), GrouplessAPIPrefixes: sets.NewString("api")} requestInfoFactory := &apirequest.RequestInfoFactory{APIPrefixes: sets.NewString("apis", "api"), GrouplessAPIPrefixes: sets.NewString("api")}
longRunningRequestCheck := BasicLongRunningRequestCheck(sets.NewString("watch"), sets.NewString("proxy")) longRunningRequestCheck := BasicLongRunningRequestCheck(sets.NewString("watch"), sets.NewString("proxy"))
apfHandler := WithPriorityAndFairness(handler, longRunningRequestCheck, filter, defaultRequestWidthEstimator) apfHandler := WithPriorityAndFairness(handler, longRunningRequestCheck, filter, defaultRequestWorkEstimator)
// add the handler in the chain that adds the specified user to the request context // add the handler in the chain that adds the specified user to the request context
handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

8
pkg/util/flowcontrol/apf_controller.go
View File

@ -80,9 +80,9 @@ type StartFunction func(ctx context.Context, hashValue uint64) (execute bool, af
// RequestDigest holds necessary info from request for flow-control // RequestDigest holds necessary info from request for flow-control
type RequestDigest struct { type RequestDigest struct {
RequestInfo *request.RequestInfo RequestInfo *request.RequestInfo
User user.Info User user.Info
Width fcrequest.Width WorkEstimate fcrequest.WorkEstimate
} }
// `*configController` maintains eventual consistency with the API // `*configController` maintains eventual consistency with the API
@ -809,7 +809,7 @@ func (cfgCtlr *configController) startRequest(ctx context.Context, rd RequestDig
} }
startWaitingTime = time.Now() startWaitingTime = time.Now()
klog.V(7).Infof("startRequest(%#+v) => fsName=%q, distMethod=%#+v, plName=%q, numQueues=%d", rd, selectedFlowSchema.Name, selectedFlowSchema.Spec.DistinguisherMethod, plName, numQueues) klog.V(7).Infof("startRequest(%#+v) => fsName=%q, distMethod=%#+v, plName=%q, numQueues=%d", rd, selectedFlowSchema.Name, selectedFlowSchema.Spec.DistinguisherMethod, plName, numQueues)
req, idle := plState.queues.StartRequest(ctx, &rd.Width, hashValue, flowDistinguisher, selectedFlowSchema.Name, rd.RequestInfo, rd.User, queueNoteFn) req, idle := plState.queues.StartRequest(ctx, &rd.WorkEstimate, hashValue, flowDistinguisher, selectedFlowSchema.Name, rd.RequestInfo, rd.User, queueNoteFn)
if idle { if idle {
cfgCtlr.maybeReapLocked(plName, plState) cfgCtlr.maybeReapLocked(plName, plState)
} }

2
pkg/util/flowcontrol/controller_test.go
View File

@ -140,7 +140,7 @@ func (cqs *ctlrTestQueueSet) IsIdle() bool {
return cqs.countActive == 0 return cqs.countActive == 0
} }
func (cqs *ctlrTestQueueSet) StartRequest(ctx context.Context, width *fcrequest.Width, hashValue uint64, flowDistinguisher, fsName string, descr1, descr2 interface{}, queueNoteFn fq.QueueNoteFn) (req fq.Request, idle bool) { func (cqs *ctlrTestQueueSet) StartRequest(ctx context.Context, width *fcrequest.WorkEstimate, hashValue uint64, flowDistinguisher, fsName string, descr1, descr2 interface{}, queueNoteFn fq.QueueNoteFn) (req fq.Request, idle bool) {
cqs.cts.lock.Lock() cqs.cts.lock.Lock()
defer cqs.cts.lock.Unlock() defer cqs.cts.lock.Unlock()
cqs.countActive++ cqs.countActive++

2
pkg/util/flowcontrol/fairqueuing/interface.go
View File

@ -81,7 +81,7 @@ type QueueSet interface {
// was idle at the moment of the return. Otherwise idle==false // was idle at the moment of the return. Otherwise idle==false
// and the client must call the Finish method of the Request // and the client must call the Finish method of the Request
// exactly once. // exactly once.
StartRequest(ctx context.Context, width *request.Width, hashValue uint64, flowDistinguisher, fsName string, descr1, descr2 interface{}, queueNoteFn QueueNoteFn) (req Request, idle bool) StartRequest(ctx context.Context, width *request.WorkEstimate, hashValue uint64, flowDistinguisher, fsName string, descr1, descr2 interface{}, queueNoteFn QueueNoteFn) (req Request, idle bool)
// UpdateObservations makes sure any time-based statistics have // UpdateObservations makes sure any time-based statistics have
// caught up with the current clock reading // caught up with the current clock reading

2
pkg/util/flowcontrol/fairqueuing/queueset/fifo_list_test.go
View File

@ -154,7 +154,7 @@ func TestFIFOSeatsSum(t *testing.T) {
list := newRequestFIFO() list := newRequestFIFO()
newRequest := func(width uint) *request { newRequest := func(width uint) *request {
return &request{width: fcrequest.Width{Seats: width}} return &request{workEstimate: fcrequest.WorkEstimate{Seats: width}}
} }
arrival := []*request{newRequest(1), newRequest(2), newRequest(3)} arrival := []*request{newRequest(1), newRequest(2), newRequest(3)}
removeFn := make([]removeFromFIFOFunc, 0) removeFn := make([]removeFromFIFOFunc, 0)

22
pkg/util/flowcontrol/fairqueuing/queueset/queueset.go
View File

@ -235,7 +235,7 @@ const (
// executing at each point where there is a change in that quantity, // executing at each point where there is a change in that quantity,
// because the metrics --- and only the metrics --- track that // because the metrics --- and only the metrics --- track that
// quantity per FlowSchema. // quantity per FlowSchema.
func (qs *queueSet) StartRequest(ctx context.Context, width *fqrequest.Width, hashValue uint64, flowDistinguisher, fsName string, descr1, descr2 interface{}, queueNoteFn fq.QueueNoteFn) (fq.Request, bool) { func (qs *queueSet) StartRequest(ctx context.Context, workEstimate *fqrequest.WorkEstimate, hashValue uint64, flowDistinguisher, fsName string, descr1, descr2 interface{}, queueNoteFn fq.QueueNoteFn) (fq.Request, bool) {
qs.lockAndSyncTime() qs.lockAndSyncTime()
defer qs.lock.Unlock() defer qs.lock.Unlock()
var req *request var req *request
@ -244,13 +244,13 @@ func (qs *queueSet) StartRequest(ctx context.Context, width *fqrequest.Width, ha
// Step 0: // Step 0:
// Apply only concurrency limit, if zero queues desired // Apply only concurrency limit, if zero queues desired
if qs.qCfg.DesiredNumQueues < 1 { if qs.qCfg.DesiredNumQueues < 1 {
if !qs.canAccommodateSeatsLocked(int(width.Seats)) { if !qs.canAccommodateSeatsLocked(int(workEstimate.Seats)) {
klog.V(5).Infof("QS(%s): rejecting request %q %#+v %#+v because %d seats are asked for, %d seats are in use (%d are executing) and the limit is %d", klog.V(5).Infof("QS(%s): rejecting request %q %#+v %#+v because %d seats are asked for, %d seats are in use (%d are executing) and the limit is %d",
qs.qCfg.Name, fsName, descr1, descr2, width, qs.totSeatsInUse, qs.totRequestsExecuting, qs.dCfg.ConcurrencyLimit) qs.qCfg.Name, fsName, descr1, descr2, workEstimate, qs.totSeatsInUse, qs.totRequestsExecuting, qs.dCfg.ConcurrencyLimit)
metrics.AddReject(ctx, qs.qCfg.Name, fsName, "concurrency-limit") metrics.AddReject(ctx, qs.qCfg.Name, fsName, "concurrency-limit")
return nil, qs.isIdleLocked() return nil, qs.isIdleLocked()
} }
req = qs.dispatchSansQueueLocked(ctx, width, flowDistinguisher, fsName, descr1, descr2) req = qs.dispatchSansQueueLocked(ctx, workEstimate, flowDistinguisher, fsName, descr1, descr2)
return req, false return req, false
} }
@ -261,7 +261,7 @@ func (qs *queueSet) StartRequest(ctx context.Context, width *fqrequest.Width, ha
// 3) Reject current request if there is not enough concurrency shares and // 3) Reject current request if there is not enough concurrency shares and
// we are at max queue length // we are at max queue length
// 4) If not rejected, create a request and enqueue // 4) If not rejected, create a request and enqueue
req = qs.timeoutOldRequestsAndRejectOrEnqueueLocked(ctx, width, hashValue, flowDistinguisher, fsName, descr1, descr2, queueNoteFn) req = qs.timeoutOldRequestsAndRejectOrEnqueueLocked(ctx, workEstimate, hashValue, flowDistinguisher, fsName, descr1, descr2, queueNoteFn)
// req == nil means that the request was rejected - no remaining // req == nil means that the request was rejected - no remaining
// concurrency shares and at max queue length already // concurrency shares and at max queue length already
if req == nil { if req == nil {
@ -316,7 +316,7 @@ func (qs *queueSet) StartRequest(ctx context.Context, width *fqrequest.Width, ha
// Seats returns the number of seats this request requires. // Seats returns the number of seats this request requires.
func (req *request) Seats() int { func (req *request) Seats() int {
return int(req.width.Seats) return int(req.workEstimate.Seats)
} }
func (req *request) NoteQueued(inQueue bool) { func (req *request) NoteQueued(inQueue bool) {
@ -437,7 +437,7 @@ func (qs *queueSet) getVirtualTimeRatioLocked() float64 {
// returns the enqueud request on a successful enqueue // returns the enqueud request on a successful enqueue
// returns nil in the case that there is no available concurrency or // returns nil in the case that there is no available concurrency or
// the queuelengthlimit has been reached // the queuelengthlimit has been reached
func (qs *queueSet) timeoutOldRequestsAndRejectOrEnqueueLocked(ctx context.Context, width *fqrequest.Width, hashValue uint64, flowDistinguisher, fsName string, descr1, descr2 interface{}, queueNoteFn fq.QueueNoteFn) *request { func (qs *queueSet) timeoutOldRequestsAndRejectOrEnqueueLocked(ctx context.Context, workEstimate *fqrequest.WorkEstimate, hashValue uint64, flowDistinguisher, fsName string, descr1, descr2 interface{}, queueNoteFn fq.QueueNoteFn) *request {
// Start with the shuffle sharding, to pick a queue. // Start with the shuffle sharding, to pick a queue.
queueIdx := qs.chooseQueueIndexLocked(hashValue, descr1, descr2) queueIdx := qs.chooseQueueIndexLocked(hashValue, descr1, descr2)
queue := qs.queues[queueIdx] queue := qs.queues[queueIdx]
@ -459,7 +459,7 @@ func (qs *queueSet) timeoutOldRequestsAndRejectOrEnqueueLocked(ctx context.Conte
descr1: descr1, descr1: descr1,
descr2: descr2, descr2: descr2,
queueNoteFn: queueNoteFn, queueNoteFn: queueNoteFn,
width: *width, workEstimate: *workEstimate,
} }
if ok := qs.rejectOrEnqueueLocked(req); !ok { if ok := qs.rejectOrEnqueueLocked(req); !ok {
return nil return nil
@ -476,7 +476,7 @@ func (qs *queueSet) chooseQueueIndexLocked(hashValue uint64, descr1, descr2 inte
// the dealer uses the current desired number of queues, which is no larger than the number in `qs.queues`. // the dealer uses the current desired number of queues, which is no larger than the number in `qs.queues`.
qs.dealer.Deal(hashValue, func(queueIdx int) { qs.dealer.Deal(hashValue, func(queueIdx int) {
// TODO: Consider taking into account `additional latency` of requests // TODO: Consider taking into account `additional latency` of requests
// in addition to their widths. // in addition to their seats.
// Ideally, this should be based on projected completion time in the // Ideally, this should be based on projected completion time in the
// virtual world of the youngest request in the queue. // virtual world of the youngest request in the queue.
thisSeatsSum := qs.queues[queueIdx].requests.SeatsSum() thisSeatsSum := qs.queues[queueIdx].requests.SeatsSum()
@ -579,7 +579,7 @@ func (qs *queueSet) dispatchAsMuchAsPossibleLocked() {
} }
} }
func (qs *queueSet) dispatchSansQueueLocked(ctx context.Context, width *fqrequest.Width, flowDistinguisher, fsName string, descr1, descr2 interface{}) *request { func (qs *queueSet) dispatchSansQueueLocked(ctx context.Context, workEstimate *fqrequest.WorkEstimate, flowDistinguisher, fsName string, descr1, descr2 interface{}) *request {
// does not call metrics.SetDispatchMetrics because there is no queuing and thus no interesting virtual world // does not call metrics.SetDispatchMetrics because there is no queuing and thus no interesting virtual world
now := qs.clock.Now() now := qs.clock.Now()
req := &request{ req := &request{
@ -592,7 +592,7 @@ func (qs *queueSet) dispatchSansQueueLocked(ctx context.Context, width *fqreques
arrivalTime: now, arrivalTime: now,
descr1: descr1, descr1: descr1,
descr2: descr2, descr2: descr2,
width: *width, workEstimate: *workEstimate,
} }
req.decision.SetLocked(decisionExecute) req.decision.SetLocked(decisionExecute)
qs.totRequestsExecuting++ qs.totRequestsExecuting++

26
pkg/util/flowcontrol/fairqueuing/queueset/queueset_test.go
View File

@ -227,7 +227,7 @@ func (ust *uniformScenarioThread) callK(k int) {
if k >= ust.nCalls { if k >= ust.nCalls {
return return
} }
req, idle := ust.uss.qs.StartRequest(context.Background(), &fcrequest.Width{Seats: 1}, ust.uc.hash, "", ust.fsName, ust.uss.name, []int{ust.i, ust.j, k}, nil) req, idle := ust.uss.qs.StartRequest(context.Background(), &fcrequest.WorkEstimate{Seats: 1}, ust.uc.hash, "", ust.fsName, ust.uss.name, []int{ust.i, ust.j, k}, nil)
ust.uss.t.Logf("%s: %d, %d, %d got req=%p, idle=%v", ust.uss.clk.Now().Format(nsTimeFmt), ust.i, ust.j, k, req, idle) ust.uss.t.Logf("%s: %d, %d, %d got req=%p, idle=%v", ust.uss.clk.Now().Format(nsTimeFmt), ust.i, ust.j, k, req, idle)
if req == nil { if req == nil {
atomic.AddUint64(&ust.uss.failedCount, 1) atomic.AddUint64(&ust.uss.failedCount, 1)
@ -672,7 +672,7 @@ func TestContextCancel(t *testing.T) {
ctx1 := context.Background() ctx1 := context.Background()
b2i := map[bool]int{false: 0, true: 1} b2i := map[bool]int{false: 0, true: 1}
var qnc [2][2]int32 var qnc [2][2]int32
req1, _ := qs.StartRequest(ctx1, &fcrequest.Width{Seats: 1}, 1, "", "fs1", "test", "one", func(inQueue bool) { atomic.AddInt32(&qnc[0][b2i[inQueue]], 1) }) req1, _ := qs.StartRequest(ctx1, &fcrequest.WorkEstimate{Seats: 1}, 1, "", "fs1", "test", "one", func(inQueue bool) { atomic.AddInt32(&qnc[0][b2i[inQueue]], 1) })
if req1 == nil { if req1 == nil {
t.Error("Request rejected") t.Error("Request rejected")
return return
@ -700,7 +700,7 @@ func TestContextCancel(t *testing.T) {
counter.Add(1) counter.Add(1)
cancel2() cancel2()
}() }()
req2, idle2a := qs.StartRequest(ctx2, &fcrequest.Width{Seats: 1}, 2, "", "fs2", "test", "two", func(inQueue bool) { atomic.AddInt32(&qnc[1][b2i[inQueue]], 1) }) req2, idle2a := qs.StartRequest(ctx2, &fcrequest.WorkEstimate{Seats: 1}, 2, "", "fs2", "test", "two", func(inQueue bool) { atomic.AddInt32(&qnc[1][b2i[inQueue]], 1) })
if idle2a { if idle2a {
t.Error("2nd StartRequest returned idle") t.Error("2nd StartRequest returned idle")
} }
@ -759,7 +759,7 @@ func TestTotalRequestsExecutingWithPanic(t *testing.T) {
} }
ctx := context.Background() ctx := context.Background()
req, _ := qs.StartRequest(ctx, &fcrequest.Width{Seats: 1}, 1, "", "fs", "test", "one", func(inQueue bool) {}) req, _ := qs.StartRequest(ctx, &fcrequest.WorkEstimate{Seats: 1}, 1, "", "fs", "test", "one", func(inQueue bool) {})
if req == nil { if req == nil {
t.Fatal("expected a Request object from StartRequest, but got nil") t.Fatal("expected a Request object from StartRequest, but got nil")
} }
@ -812,13 +812,13 @@ func TestSelectQueueLocked(t *testing.T) {
{ {
virtualStart: 200, virtualStart: 200,
requests: newFIFO( requests: newFIFO(
&request{width: fcrequest.Width{Seats: 1}}, &request{workEstimate: fcrequest.WorkEstimate{Seats: 1}},
), ),
}, },
{ {
virtualStart: 100, virtualStart: 100,
requests: newFIFO( requests: newFIFO(
&request{width: fcrequest.Width{Seats: 1}}, &request{workEstimate: fcrequest.WorkEstimate{Seats: 1}},
), ),
}, },
}, },
@ -835,7 +835,7 @@ func TestSelectQueueLocked(t *testing.T) {
{ {
virtualStart: 200, virtualStart: 200,
requests: newFIFO( requests: newFIFO(
&request{width: fcrequest.Width{Seats: 1}}, &request{workEstimate: fcrequest.WorkEstimate{Seats: 1}},
), ),
}, },
}, },
@ -852,13 +852,13 @@ func TestSelectQueueLocked(t *testing.T) {
{ {
virtualStart: 200, virtualStart: 200,
requests: newFIFO( requests: newFIFO(
&request{width: fcrequest.Width{Seats: 50}}, &request{workEstimate: fcrequest.WorkEstimate{Seats: 50}},
), ),
}, },
{ {
virtualStart: 100, virtualStart: 100,
requests: newFIFO( requests: newFIFO(
&request{width: fcrequest.Width{Seats: 25}}, &request{workEstimate: fcrequest.WorkEstimate{Seats: 25}},
), ),
}, },
}, },
@ -875,13 +875,13 @@ func TestSelectQueueLocked(t *testing.T) {
{ {
virtualStart: 200, virtualStart: 200,
requests: newFIFO( requests: newFIFO(
&request{width: fcrequest.Width{Seats: 10}}, &request{workEstimate: fcrequest.WorkEstimate{Seats: 10}},
), ),
}, },
{ {
virtualStart: 100, virtualStart: 100,
requests: newFIFO( requests: newFIFO(
&request{width: fcrequest.Width{Seats: 25}}, &request{workEstimate: fcrequest.WorkEstimate{Seats: 25}},
), ),
}, },
}, },
@ -898,13 +898,13 @@ func TestSelectQueueLocked(t *testing.T) {
{ {
virtualStart: 200, virtualStart: 200,
requests: newFIFO( requests: newFIFO(
&request{width: fcrequest.Width{Seats: 10}}, &request{workEstimate: fcrequest.WorkEstimate{Seats: 10}},
), ),
}, },
{ {
virtualStart: 100, virtualStart: 100,
requests: newFIFO( requests: newFIFO(
&request{width: fcrequest.Width{Seats: 25}}, &request{workEstimate: fcrequest.WorkEstimate{Seats: 25}},
), ),
}, },
}, },

4
pkg/util/flowcontrol/fairqueuing/queueset/types.go
View File

@ -44,8 +44,8 @@ type request struct {
// startTime is the real time when the request began executing // startTime is the real time when the request began executing
startTime time.Time startTime time.Time
// width of the request // estimated amount of work of the request
width fcrequest.Width workEstimate fcrequest.WorkEstimate
// decision gets set to a `requestDecision` indicating what to do // decision gets set to a `requestDecision` indicating what to do
// with this request. It gets set exactly once, when the request // with this request. It gets set exactly once, when the request

2
pkg/util/flowcontrol/fairqueuing/testing/no-restraint.go
View File

@ -56,7 +56,7 @@ func (noRestraint) IsIdle() bool {
return false return false
} }
func (noRestraint) StartRequest(ctx context.Context, width *fcrequest.Width, hashValue uint64, flowDistinguisher, fsName string, descr1, descr2 interface{}, queueNoteFn fq.QueueNoteFn) (fq.Request, bool) { func (noRestraint) StartRequest(ctx context.Context, workEstimate *fcrequest.WorkEstimate, hashValue uint64, flowDistinguisher, fsName string, descr1, descr2 interface{}, queueNoteFn fq.QueueNoteFn) (fq.Request, bool) {
return noRestraintRequest{}, false return noRestraintRequest{}, false
} }

12
pkg/util/flowcontrol/match_test.go
View File

@ -102,8 +102,8 @@ func TestLiterals(t *testing.T) {
Name: "eman", Name: "eman",
Parts: []string{"goodrscs", "eman"}, Parts: []string{"goodrscs", "eman"},
}, },
User: ui, User: ui,
Width: fcrequest.Width{Seats: 1}, WorkEstimate: fcrequest.WorkEstimate{Seats: 1},
} }
reqRU := RequestDigest{ reqRU := RequestDigest{
RequestInfo: &request.RequestInfo{ RequestInfo: &request.RequestInfo{
@ -118,8 +118,8 @@ func TestLiterals(t *testing.T) {
Name: "eman", Name: "eman",
Parts: []string{"goodrscs", "eman"}, Parts: []string{"goodrscs", "eman"},
}, },
User: ui, User: ui,
Width: fcrequest.Width{Seats: 1}, WorkEstimate: fcrequest.WorkEstimate{Seats: 1},
} }
reqN := RequestDigest{ reqN := RequestDigest{
RequestInfo: &request.RequestInfo{ RequestInfo: &request.RequestInfo{
@ -127,8 +127,8 @@ func TestLiterals(t *testing.T) {
Path: "/openapi/v2", Path: "/openapi/v2",
Verb: "goodverb", Verb: "goodverb",
}, },
User: ui, User: ui,
Width: fcrequest.Width{Seats: 1}, WorkEstimate: fcrequest.WorkEstimate{Seats: 1},
} }
checkRules(t, true, reqRN, []flowcontrol.PolicyRulesWithSubjects{{ checkRules(t, true, reqRN, []flowcontrol.PolicyRulesWithSubjects{{
Subjects: []flowcontrol.Subject{{Kind: flowcontrol.SubjectKindUser, Subjects: []flowcontrol.Subject{{Kind: flowcontrol.SubjectKindUser,

22
pkg/util/flowcontrol/request/width.go
View File

@ -20,28 +20,28 @@ import (
"net/http" "net/http"
) )
type Width struct { type WorkEstimate struct {
// Seats represents the number of seats associated with this request // Seats represents the number of seats associated with this request
Seats uint Seats uint
} }
// DefaultWidthEstimator returns returns '1' as the "width" // DefaultWorkEstimator returns estimation with default number of seats
// of the given request. // of 1.
// //
// TODO: when we plumb in actual "width" handling for different // TODO: when we plumb in actual work estimate handling for different
// type of request(s) this function will iterate through a chain // type of request(s) this function will iterate through a chain
// of widthEstimator instance(s). // of workEstimator instance(s).
func DefaultWidthEstimator(_ *http.Request) Width { func DefaultWorkEstimator(_ *http.Request) WorkEstimate {
return Width{ return WorkEstimate{
Seats: 1, Seats: 1,
} }
} }
// WidthEstimatorFunc returns the estimated "width" of a given request. // WorkEstimatorFunc returns the estimated work of a given request.
// This function will be used by the Priority & Fairness filter to // This function will be used by the Priority & Fairness filter to
// estimate the "width" of incoming requests. // estimate the work of of incoming requests.
type WidthEstimatorFunc func(*http.Request) Width type WorkEstimatorFunc func(*http.Request) WorkEstimate
func (e WidthEstimatorFunc) EstimateWidth(r *http.Request) Width { func (e WorkEstimatorFunc) EstimateWork(r *http.Request) WorkEstimate {
return e(r) return e(r)
} }