1/* 2Copyright 2019 The Kubernetes Authors. 3 4Licensed under the Apache License, Version 2.0 (the "License"); 5you may not use this file except in compliance with the License. 6You may obtain a copy of the License at 7 8 http://www.apache.org/licenses/LICENSE-2.0 9 10Unless required by applicable law or agreed to in writing, software 11distributed under the License is distributed on an "AS IS" BASIS, 12WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13See the License for the specific language governing permissions and 14limitations under the License. 15*/ 16 17package fairqueuing 18 19import ( 20 "context" 21 "time" 22 23 "k8s.io/apiserver/pkg/util/flowcontrol/debug" 24 "k8s.io/apiserver/pkg/util/flowcontrol/metrics" 25) 26 27// QueueSetFactory is used to create QueueSet objects. Creation, like 28// config update, is done in two phases: the first phase consumes the 29// QueuingConfig and the second consumes the DispatchingConfig. They 30// are separated so that errors from the first phase can be found 31// before committing to a concurrency allotment for the second. 32type QueueSetFactory interface { 33 // BeginConstruction does the first phase of creating a QueueSet 34 BeginConstruction(QueuingConfig, metrics.TimedObserverPair) (QueueSetCompleter, error) 35} 36 37// QueueSetCompleter finishes the two-step process of creating or 38// reconfiguring a QueueSet 39type QueueSetCompleter interface { 40 // Complete returns a QueueSet configured by the given 41 // dispatching configuration. 42 Complete(DispatchingConfig) QueueSet 43} 44 45// QueueSet is the abstraction for the queuing and dispatching 46// functionality of one non-exempt priority level. It covers the 47// functionality described in the "Assignment to a Queue", "Queuing", 48// and "Dispatching" sections of 49// https://github.com/kubernetes/enhancements/blob/master/keps/sig-api-machinery/20190228-priority-and-fairness.md 50// . Some day we may have connections between priority levels, but 51// today is not that day. 52type QueueSet interface { 53 // BeginConfigChange starts the two-step process of updating the 54 // configuration. No change is made until Complete is called. If 55 // `C := X.BeginConstruction(q)` then `C.Complete(d)` returns the 56 // same value `X`. If the QueuingConfig's DesiredNumQueues field 57 // is zero then the other queuing-specific config parameters are 58 // not changed, so that the queues continue draining as before. 59 // In any case, reconfiguration does not discard any queue unless 60 // and until it is undesired and empty. 61 BeginConfigChange(QueuingConfig) (QueueSetCompleter, error) 62 63 // IsIdle returns a bool indicating whether the QueueSet was idle 64 // at the moment of the return. Idle means the QueueSet has zero 65 // requests queued and zero executing. This bit can change only 66 // (1) during a call to StartRequest and (2) during a call to 67 // Request::Finish. In the latter case idleness can only change 68 // from false to true. 69 IsIdle() bool 70 71 // StartRequest begins the process of handling a request. If the 72 // request gets queued and the number of queues is greater than 1 73 // then StartRequest uses the given hashValue as the source of 74 // entropy as it shuffle-shards the request into a queue. The 75 // descr1 and descr2 values play no role in the logic but appear 76 // in log messages. This method always returns quickly (without 77 // waiting for the request to be dequeued). If this method 78 // returns a nil Request value then caller should reject the 79 // request and the returned bool indicates whether the QueueSet 80 // was idle at the moment of the return. Otherwise idle==false 81 // and the client must call the Finish method of the Request 82 // exactly once. 83 StartRequest(ctx context.Context, hashValue uint64, flowDistinguisher, fsName string, descr1, descr2 interface{}, queueNoteFn QueueNoteFn) (req Request, idle bool) 84 85 // UpdateObservations makes sure any time-based statistics have 86 // caught up with the current clock reading 87 UpdateObservations() 88 89 // Dump saves and returns the instant internal state of the queue-set. 90 // Note that dumping process will stop the queue-set from proceeding 91 // any requests. 92 // For debugging only. 93 Dump(includeRequestDetails bool) debug.QueueSetDump 94} 95 96// QueueNoteFn is called when a request enters and leaves a queue 97type QueueNoteFn func(inQueue bool) 98 99// Request represents the remainder of the handling of one request 100type Request interface { 101 // Finish determines whether to execute or reject the request and 102 // invokes `execute` if the decision is to execute the request. 103 // The returned `idle bool` value indicates whether the QueueSet 104 // was idle when the value was calculated, but might no longer be 105 // accurate by the time the client examines that value. 106 Finish(execute func()) (idle bool) 107} 108 109// QueuingConfig defines the configuration of the queuing aspect of a QueueSet. 110type QueuingConfig struct { 111 // Name is used to identify a queue set, allowing for descriptive information about its intended use 112 Name string 113 114 // DesiredNumQueues is the number of queues that the API says 115 // should exist now. This may be zero, in which case 116 // QueueLengthLimit, HandSize, and RequestWaitLimit are ignored. 117 DesiredNumQueues int 118 119 // QueueLengthLimit is the maximum number of requests that may be waiting in a given queue at a time 120 QueueLengthLimit int 121 122 // HandSize is a parameter of shuffle sharding. Upon arrival of a request, a queue is chosen by randomly 123 // dealing a "hand" of this many queues and then picking one of minimum length. 124 HandSize int 125 126 // RequestWaitLimit is the maximum amount of time that a request may wait in a queue. 127 // If, by the end of that time, the request has not been dispatched then it is rejected. 128 RequestWaitLimit time.Duration 129} 130 131// DispatchingConfig defines the configuration of the dispatching aspect of a QueueSet. 132type DispatchingConfig struct { 133 // ConcurrencyLimit is the maximum number of requests of this QueueSet that may be executing at a time 134 ConcurrencyLimit int 135} 136