|
| 1 | +package expectation |
| 2 | + |
| 3 | +import ( |
| 4 | + "fmt" |
| 5 | + log "github.com/sirupsen/logrus" |
| 6 | + "sync/atomic" |
| 7 | + "time" |
| 8 | + |
| 9 | + "k8s.io/apimachinery/pkg/util/clock" |
| 10 | + "k8s.io/client-go/tools/cache" |
| 11 | +) |
| 12 | + |
| 13 | +const ( |
| 14 | + // If a watch drops a delete event for a pod, it'll take this long |
| 15 | + // before a dormant controller waiting for those packets is woken up anyway. It is |
| 16 | + // specifically targeted at the case where some problem prevents an update |
| 17 | + // of expectations, without it the controller could stay asleep forever. This should |
| 18 | + // be set based on the expected latency of watch events. |
| 19 | + // |
| 20 | + // Currently a controller can service (create *and* observe the watch events for said |
| 21 | + // creation) about 10 pods a second, so it takes about 1 min to service |
| 22 | + // 500 pods. Just creation is limited to 20qps, and watching happens with ~10-30s |
| 23 | + // latency/pod at the scale of 3000 pods over 100 nodes. |
| 24 | + ExpectationsTimeout = 5 * time.Minute |
| 25 | +) |
| 26 | + |
| 27 | +// Expectations are a way for controllers to tell the controller manager what they expect. eg: |
| 28 | +// ControllerExpectations: { |
| 29 | +// controller1: expects 2 adds in 2 minutes |
| 30 | +// controller2: expects 2 dels in 2 minutes |
| 31 | +// controller3: expects -1 adds in 2 minutes => controller3's expectations have already been met |
| 32 | +// } |
| 33 | +// |
| 34 | +// Implementation: |
| 35 | +// ControlleeExpectation = pair of atomic counters to track controllee's creation/deletion |
| 36 | +// ControllerExpectationsStore = TTLStore + a ControlleeExpectation per controller |
| 37 | +// |
| 38 | +// * Once set expectations can only be lowered |
| 39 | +// * A controller isn't synced till its expectations are either fulfilled, or expire |
| 40 | +// * Controllers that don't set expectations will get woken up for every matching controllee |
| 41 | + |
| 42 | +// ExpKeyFunc to parse out the key from a ControlleeExpectation |
| 43 | +var ExpKeyFunc = func(obj interface{}) (string, error) { |
| 44 | + if e, ok := obj.(*ControlleeExpectations); ok { |
| 45 | + return e.key, nil |
| 46 | + } |
| 47 | + return "", fmt.Errorf("could not find key for obj %#v", obj) |
| 48 | +} |
| 49 | + |
| 50 | +// ControllerExpectationsInterface is an interface that allows users to set and wait on expectations. |
| 51 | +// Only abstracted out for testing. |
| 52 | +// Warning: if using KeyFunc it is not safe to use a single ControllerExpectationsInterface with different |
| 53 | +// types of controllers, because the keys might conflict across types. |
| 54 | +type ControllerExpectationsInterface interface { |
| 55 | + GetExpectations(controllerKey string) (*ControlleeExpectations, bool, error) |
| 56 | + SatisfiedExpectations(controllerKey string) bool |
| 57 | + DeleteExpectations(controllerKey string) |
| 58 | + SetExpectations(controllerKey string, add, del int) error |
| 59 | + ExpectCreations(controllerKey string, adds int) error |
| 60 | + ExpectDeletions(controllerKey string, dels int) error |
| 61 | + CreationObserved(controllerKey string) |
| 62 | + DeletionObserved(controllerKey string) |
| 63 | + RaiseExpectations(controllerKey string, add, del int) |
| 64 | + LowerExpectations(controllerKey string, add, del int) |
| 65 | +} |
| 66 | + |
| 67 | +// ControllerExpectations is a cache mapping controllers to what they expect to see before being woken up for a sync. |
| 68 | +type ControllerExpectations struct { |
| 69 | + cache.Store |
| 70 | +} |
| 71 | + |
| 72 | +// GetExpectations returns the ControlleeExpectations of the given controller. |
| 73 | +func (r *ControllerExpectations) GetExpectations(controllerKey string) (*ControlleeExpectations, bool, error) { |
| 74 | + exp, exists, err := r.GetByKey(controllerKey) |
| 75 | + if err == nil && exists { |
| 76 | + return exp.(*ControlleeExpectations), true, nil |
| 77 | + } |
| 78 | + return nil, false, err |
| 79 | +} |
| 80 | + |
| 81 | +// DeleteExpectations deletes the expectations of the given controller from the TTLStore. |
| 82 | +func (r *ControllerExpectations) DeleteExpectations(controllerKey string) { |
| 83 | + if exp, exists, err := r.GetByKey(controllerKey); err == nil && exists { |
| 84 | + if err := r.Delete(exp); err != nil { |
| 85 | + log.Infof("Error deleting expectations for controller %v: %v", controllerKey, err) |
| 86 | + } |
| 87 | + } |
| 88 | +} |
| 89 | + |
| 90 | +// SatisfiedExpectations returns true if the required adds/dels for the given controller have been observed. |
| 91 | +// Add/del counts are established by the controller at sync time, and updated as controllees are observed by the controller |
| 92 | +// manager. |
| 93 | +func (r *ControllerExpectations) SatisfiedExpectations(controllerKey string) bool { |
| 94 | + if exp, exists, err := r.GetExpectations(controllerKey); exists { |
| 95 | + if exp.Fulfilled() { |
| 96 | + log.Infof("Controller expectations fulfilled %#v", exp) |
| 97 | + return true |
| 98 | + } else if exp.isExpired() { |
| 99 | + log.Infof("Controller expectations expired %#v", exp) |
| 100 | + return true |
| 101 | + } else { |
| 102 | + log.Infof("Controller still waiting on expectations %#v", exp) |
| 103 | + return false |
| 104 | + } |
| 105 | + } else if err != nil { |
| 106 | + log.Infof("Error encountered while checking expectations %#v, forcing sync", err) |
| 107 | + } else { |
| 108 | + // When a new controller is created, it doesn't have expectations. |
| 109 | + // When it doesn't see expected watch events for > TTL, the expectations expire. |
| 110 | + // - In this case it wakes up, creates/deletes controllees, and sets expectations again. |
| 111 | + // When it has satisfied expectations and no controllees need to be created/destroyed > TTL, the expectations expire. |
| 112 | + // - In this case it continues without setting expectations till it needs to create/delete controllees. |
| 113 | + log.Infof("Controller %v either never recorded expectations, or the ttl expired.", controllerKey) |
| 114 | + } |
| 115 | + // Trigger a sync if we either encountered and error (which shouldn't happen since we're |
| 116 | + // getting from local store) or this controller hasn't established expectations. |
| 117 | + return true |
| 118 | +} |
| 119 | + |
| 120 | +// TODO: Extend ExpirationCache to support explicit expiration. |
| 121 | +// TODO: Make this possible to disable in tests. |
| 122 | +// TODO: Support injection of clock. |
| 123 | +func (exp *ControlleeExpectations) isExpired() bool { |
| 124 | + return clock.RealClock{}.Since(exp.timestamp) > ExpectationsTimeout |
| 125 | +} |
| 126 | + |
| 127 | +// SetExpectations registers new expectations for the given controller. Forgets existing expectations. |
| 128 | +func (r *ControllerExpectations) SetExpectations(controllerKey string, add, del int) error { |
| 129 | + exp := &ControlleeExpectations{add: int64(add), del: int64(del), key: controllerKey, timestamp: clock.RealClock{}.Now()} |
| 130 | + log.Infof("Setting expectations %#v", exp) |
| 131 | + return r.Add(exp) |
| 132 | +} |
| 133 | + |
| 134 | +func (r *ControllerExpectations) ExpectCreations(controllerKey string, adds int) error { |
| 135 | + return r.SetExpectations(controllerKey, adds, 0) |
| 136 | +} |
| 137 | + |
| 138 | +func (r *ControllerExpectations) ExpectDeletions(controllerKey string, dels int) error { |
| 139 | + return r.SetExpectations(controllerKey, 0, dels) |
| 140 | +} |
| 141 | + |
| 142 | +// Decrements the expectation counts of the given controller. |
| 143 | +func (r *ControllerExpectations) LowerExpectations(controllerKey string, add, del int) { |
| 144 | + if exp, exists, err := r.GetExpectations(controllerKey); err == nil && exists { |
| 145 | + exp.Add(int64(-add), int64(-del)) |
| 146 | + // The expectations might've been modified since the update on the previous line. |
| 147 | + log.Infof("Lowered expectations %#v", exp) |
| 148 | + } |
| 149 | +} |
| 150 | + |
| 151 | +// Increments the expectation counts of the given controller. |
| 152 | +func (r *ControllerExpectations) RaiseExpectations(controllerKey string, add, del int) { |
| 153 | + if exp, exists, err := r.GetExpectations(controllerKey); err == nil && exists { |
| 154 | + exp.Add(int64(add), int64(del)) |
| 155 | + // The expectations might've been modified since the update on the previous line. |
| 156 | + log.Infof("Raised expectations %#v", exp) |
| 157 | + } |
| 158 | +} |
| 159 | + |
| 160 | +// CreationObserved atomically decrements the `add` expectation count of the given controller. |
| 161 | +func (r *ControllerExpectations) CreationObserved(controllerKey string) { |
| 162 | + r.LowerExpectations(controllerKey, 1, 0) |
| 163 | +} |
| 164 | + |
| 165 | +// DeletionObserved atomically decrements the `del` expectation count of the given controller. |
| 166 | +func (r *ControllerExpectations) DeletionObserved(controllerKey string) { |
| 167 | + r.LowerExpectations(controllerKey, 0, 1) |
| 168 | +} |
| 169 | + |
| 170 | +// Expectations are either fulfilled, or expire naturally. |
| 171 | +type Expectations interface { |
| 172 | + Fulfilled() bool |
| 173 | +} |
| 174 | + |
| 175 | +// ControlleeExpectations track controllee creates/deletes. |
| 176 | +type ControlleeExpectations struct { |
| 177 | + // Important: Since these two int64 fields are using sync/atomic, they have to be at the top of the struct due to a bug on 32-bit platforms |
| 178 | + // See: https://golang.org/pkg/sync/atomic/ for more information |
| 179 | + add int64 |
| 180 | + del int64 |
| 181 | + key string |
| 182 | + timestamp time.Time |
| 183 | +} |
| 184 | + |
| 185 | +// Add increments the add and del counters. |
| 186 | +func (e *ControlleeExpectations) Add(add, del int64) { |
| 187 | + atomic.AddInt64(&e.add, add) |
| 188 | + atomic.AddInt64(&e.del, del) |
| 189 | +} |
| 190 | + |
| 191 | +// Fulfilled returns true if this expectation has been fulfilled. |
| 192 | +func (e *ControlleeExpectations) Fulfilled() bool { |
| 193 | + // TODO: think about why this line being atomic doesn't matter |
| 194 | + return atomic.LoadInt64(&e.add) <= 0 && atomic.LoadInt64(&e.del) <= 0 |
| 195 | +} |
| 196 | + |
| 197 | +// GetExpectations returns the add and del expectations of the controllee. |
| 198 | +func (e *ControlleeExpectations) GetExpectations() (int64, int64) { |
| 199 | + return atomic.LoadInt64(&e.add), atomic.LoadInt64(&e.del) |
| 200 | +} |
| 201 | + |
| 202 | +// NewControllerExpectations returns a store for ControllerExpectations. |
| 203 | +func NewControllerExpectations() *ControllerExpectations { |
| 204 | + return &ControllerExpectations{cache.NewStore(ExpKeyFunc)} |
| 205 | +} |
0 commit comments