1 /* 2 Copyright 2016 The Kubernetes Authors. 3 4 Licensed under the Apache License, Version 2.0 (the "License"); 5 you may not use this file except in compliance with the License. 6 You may obtain a copy of the License at 7 8 http://www.apache.org/licenses/LICENSE-2.0 9 10 Unless required by applicable law or agreed to in writing, software 11 distributed under the License is distributed on an "AS IS" BASIS, 12 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13 See the License for the specific language governing permissions and 14 limitations under the License. 15 */ 16 17 package watch 18 19 import ( 20 "context" 21 "errors" 22 "fmt" 23 "time" 24 25 "k8s.io/apimachinery/pkg/runtime" 26 "k8s.io/apimachinery/pkg/util/wait" 27 "k8s.io/apimachinery/pkg/watch" 28 "k8s.io/client-go/tools/cache" 29 "k8s.io/klog/v2" 30 ) 31 32 // PreconditionFunc returns true if the condition has been reached, false if it has not been reached yet, 33 // or an error if the condition failed or detected an error state. 34 type PreconditionFunc func(store cache.Store) (bool, error) 35 36 // ConditionFunc returns true if the condition has been reached, false if it has not been reached yet, 37 // or an error if the condition cannot be checked and should terminate. In general, it is better to define 38 // level driven conditions over edge driven conditions (pod has ready=true, vs pod modified and ready changed 39 // from false to true). 40 type ConditionFunc func(event watch.Event) (bool, error) 41 42 // ErrWatchClosed is returned when the watch channel is closed before timeout in UntilWithoutRetry. 43 var ErrWatchClosed = errors.New("watch closed before UntilWithoutRetry timeout") 44 45 // UntilWithoutRetry reads items from the watch until each provided condition succeeds, and then returns the last watch 46 // encountered. The first condition that returns an error terminates the watch (and the event is also returned). 47 // If no event has been received, the returned event will be nil. 48 // Conditions are satisfied sequentially so as to provide a useful primitive for higher level composition. 49 // Waits until context deadline or until context is canceled. 50 // 51 // Warning: Unless you have a very specific use case (probably a special Watcher) don't use this function!!! 52 // Warning: This will fail e.g. on API timeouts and/or 'too old resource version' error. 53 // Warning: You are most probably looking for a function *Until* or *UntilWithSync* below, 54 // Warning: solving such issues. 55 // TODO: Consider making this function private to prevent misuse when the other occurrences in our codebase are gone. 56 func UntilWithoutRetry(ctx context.Context, watcher watch.Interface, conditions ...ConditionFunc) (*watch.Event, error) { 57 ch := watcher.ResultChan() 58 defer watcher.Stop() 59 var lastEvent *watch.Event 60 for _, condition := range conditions { 61 // check the next condition against the previous event and short circuit waiting for the next watch 62 if lastEvent != nil { 63 done, err := condition(*lastEvent) 64 if err != nil { 65 return lastEvent, err 66 } 67 if done { 68 continue 69 } 70 } 71 ConditionSucceeded: 72 for { 73 select { 74 case event, ok := <-ch: 75 if !ok { 76 return lastEvent, ErrWatchClosed 77 } 78 lastEvent = &event 79 80 done, err := condition(event) 81 if err != nil { 82 return lastEvent, err 83 } 84 if done { 85 break ConditionSucceeded 86 } 87 88 case <-ctx.Done(): 89 return lastEvent, wait.ErrWaitTimeout 90 } 91 } 92 } 93 return lastEvent, nil 94 } 95 96 // Until wraps the watcherClient's watch function with RetryWatcher making sure that watcher gets restarted in case of errors. 97 // The initialResourceVersion will be given to watch method when first called. It shall not be "" or "0" 98 // given the underlying WATCH call issues (#74022). 99 // Remaining behaviour is identical to function UntilWithoutRetry. (See above.) 100 // Until can deal with API timeouts and lost connections. 101 // It guarantees you to see all events and in the order they happened. 102 // Due to this guarantee there is no way it can deal with 'Resource version too old error'. It will fail in this case. 103 // (See `UntilWithSync` if you'd prefer to recover from all the errors including RV too old by re-listing 104 // those items. In normal code you should care about being level driven so you'd not care about not seeing all the edges.) 105 // 106 // The most frequent usage for Until would be a test where you want to verify exact order of events ("edges"). 107 func Until(ctx context.Context, initialResourceVersion string, watcherClient cache.Watcher, conditions ...ConditionFunc) (*watch.Event, error) { 108 w, err := NewRetryWatcher(initialResourceVersion, watcherClient) 109 if err != nil { 110 return nil, err 111 } 112 113 return UntilWithoutRetry(ctx, w, conditions...) 114 } 115 116 // UntilWithSync creates an informer from lw, optionally checks precondition when the store is synced, 117 // and watches the output until each provided condition succeeds, in a way that is identical 118 // to function UntilWithoutRetry. (See above.) 119 // UntilWithSync can deal with all errors like API timeout, lost connections and 'Resource version too old'. 120 // It is the only function that can recover from 'Resource version too old', Until and UntilWithoutRetry will 121 // just fail in that case. On the other hand it can't provide you with guarantees as strong as using simple 122 // Watch method with Until. It can skip some intermediate events in case of watch function failing but it will 123 // re-list to recover and you always get an event, if there has been a change, after recovery. 124 // Also with the current implementation based on DeltaFIFO, order of the events you receive is guaranteed only for 125 // particular object, not between more of them even it's the same resource. 126 // The most frequent usage would be a command that needs to watch the "state of the world" and should't fail, like: 127 // waiting for object reaching a state, "small" controllers, ... 128 func UntilWithSync(ctx context.Context, lw cache.ListerWatcher, objType runtime.Object, precondition PreconditionFunc, conditions ...ConditionFunc) (*watch.Event, error) { 129 indexer, informer, watcher, done := NewIndexerInformerWatcher(lw, objType) 130 // We need to wait for the internal informers to fully stop so it's easier to reason about 131 // and it works with non-thread safe clients. 132 defer func() { <-done }() 133 // Proxy watcher can be stopped multiple times so it's fine to use defer here to cover alternative branches and 134 // let UntilWithoutRetry to stop it 135 defer watcher.Stop() 136 137 if precondition != nil { 138 if !cache.WaitForCacheSync(ctx.Done(), informer.HasSynced) { 139 return nil, fmt.Errorf("UntilWithSync: unable to sync caches: %w", ctx.Err()) 140 } 141 142 done, err := precondition(indexer) 143 if err != nil { 144 return nil, err 145 } 146 147 if done { 148 return nil, nil 149 } 150 } 151 152 return UntilWithoutRetry(ctx, watcher, conditions...) 153 } 154 155 // ContextWithOptionalTimeout wraps context.WithTimeout and handles infinite timeouts expressed as 0 duration. 156 func ContextWithOptionalTimeout(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc) { 157 if timeout < 0 { 158 // This should be handled in validation 159 klog.Errorf("Timeout for context shall not be negative!") 160 timeout = 0 161 } 162 163 if timeout == 0 { 164 return context.WithCancel(parent) 165 } 166 167 return context.WithTimeout(parent, timeout) 168 } 169