@@ -20,9 +20,12 @@ import (
2020 "context"
2121 "fmt"
2222 "sync"
23+ "time"
2324
2425 clientv3 "go.etcd.io/etcd/client/v3"
26+ "k8s.io/apimachinery/pkg/util/runtime"
2527 "k8s.io/apimachinery/pkg/util/version"
28+ "k8s.io/apimachinery/pkg/util/wait"
2629 "k8s.io/apiserver/pkg/storage"
2730 "k8s.io/klog/v2"
2831 "k8s.io/utils/ptr"
@@ -43,84 +46,98 @@ type FeatureSupportChecker interface {
4346 // Supports check if the feature is supported or not by checking internal cache.
4447 // By default all calls to this function before calling CheckClient returns false.
4548 // Returns true if all endpoints in etcd clients are supporting the feature.
46- Supports (feature storage.Feature ) (bool , error )
49+ // If client A supports and client B doesn't support the feature, the `Supports` will
50+ // first return true at client A initializtion and then return false on client B
51+ // initialzation, it can flip the support at runtime.
52+ Supports (feature storage.Feature ) bool
4753 // CheckClient works with etcd client to recalcualte feature support and cache it internally.
4854 // All etcd clients should support feature to cause `Supports` return true.
4955 // If client A supports and client B doesn't support the feature, the `Supports` will
5056 // first return true at client A initializtion and then return false on client B
5157 // initialzation, it can flip the support at runtime.
52- CheckClient (ctx context.Context , c client , feature storage.Feature ) error
58+ CheckClient (ctx context.Context , c client , feature storage.Feature )
5359}
5460
5561type defaultFeatureSupportChecker struct {
56- lock sync.Mutex
57- progressNotifySupported * bool
58- progresNotifyEndpointCache map [string ]bool
62+ lock sync.Mutex
63+ progressNotifySupported * bool
64+ checkingEndpoint map [string ]struct {}
5965}
6066
6167func newDefaultFeatureSupportChecker () * defaultFeatureSupportChecker {
6268 return & defaultFeatureSupportChecker {
63- progresNotifyEndpointCache : make (map [string ]bool ),
69+ checkingEndpoint : make (map [string ]struct {} ),
6470 }
6571}
6672
6773// Supports can check the featue from anywhere without storage if it was cached before.
68- func (f * defaultFeatureSupportChecker ) Supports (feature storage.Feature ) ( bool , error ) {
74+ func (f * defaultFeatureSupportChecker ) Supports (feature storage.Feature ) bool {
6975 switch feature {
7076 case storage .RequestWatchProgress :
7177 f .lock .Lock ()
7278 defer f .lock .Unlock ()
7379
74- return ptr .Deref (f .progressNotifySupported , false ), nil
80+ return ptr .Deref (f .progressNotifySupported , false )
7581 default :
76- return false , fmt .Errorf ("feature %q is not implemented in DefaultFeatureSupportChecker" , feature )
82+ runtime .HandleError (fmt .Errorf ("feature %q is not implemented in DefaultFeatureSupportChecker" , feature ))
83+ return false
7784 }
7885}
7986
8087// CheckClient accepts client and calculate the support per endpoint and caches it.
81- // It will return at any point if error happens or one endpoint is not supported.
82- func (f * defaultFeatureSupportChecker ) CheckClient (ctx context.Context , c client , feature storage.Feature ) error {
88+ func (f * defaultFeatureSupportChecker ) CheckClient (ctx context.Context , c client , feature storage.Feature ) {
8389 switch feature {
8490 case storage .RequestWatchProgress :
85- return f . clientSupportsRequestWatchProgress (ctx , c )
91+ f . checkClient (ctx , c )
8692 default :
87- return fmt .Errorf ("feature %q is not implemented in DefaultFeatureSupportChecker" , feature )
88-
93+ runtime .HandleError (fmt .Errorf ("feature %q is not implemented in DefaultFeatureSupportChecker" , feature ))
8994 }
9095}
9196
92- func (f * defaultFeatureSupportChecker ) clientSupportsRequestWatchProgress (ctx context.Context , c client ) error {
97+ func (f * defaultFeatureSupportChecker ) checkClient (ctx context.Context , c client ) {
98+ // start with 10 ms, multiply by 2 each step, until 15 s and stays on 15 seconds.
99+ delayFunc := wait.Backoff {
100+ Duration : 10 * time .Millisecond ,
101+ Cap : 15 * time .Second ,
102+ Factor : 2.0 ,
103+ Steps : 11 }.DelayFunc ()
93104 f .lock .Lock ()
94105 defer f .lock .Unlock ()
95-
96106 for _ , ep := range c .Endpoints () {
97- supported , err := f .supportsProgressNotifyEndpointLocked (ctx , c , ep )
98- if err != nil {
99- return err
107+ if _ , found := f .checkingEndpoint [ep ]; found {
108+ continue
100109 }
101- if ! supported {
102- f .progressNotifySupported = ptr .To (false )
103- return nil
104- }
105- }
106- if f .progressNotifySupported == nil && len (c .Endpoints ()) > 0 {
107- f .progressNotifySupported = ptr .To (true )
110+ f .checkingEndpoint [ep ] = struct {}{}
111+ go func (ep string ) {
112+ defer runtime .HandleCrash ()
113+ err := delayFunc .Until (ctx , true , true , func (ctx context.Context ) (done bool , err error ) {
114+ internalErr := f .clientSupportsRequestWatchProgress (ctx , c , ep )
115+ return internalErr == nil , nil
116+ })
117+ if err != nil {
118+ klog .ErrorS (err , "Failed to check if RequestWatchProgress is supported by etcd after retrying" )
119+ }
120+ }(ep )
108121 }
109- return nil
110122}
111123
112- func (f * defaultFeatureSupportChecker ) supportsProgressNotifyEndpointLocked (ctx context.Context , c client , ep string ) (bool , error ) {
113- if supported , ok := f .progresNotifyEndpointCache [ep ]; ok {
114- return supported , nil
115- }
116-
124+ func (f * defaultFeatureSupportChecker ) clientSupportsRequestWatchProgress (ctx context.Context , c client , ep string ) error {
117125 supported , err := endpointSupportsRequestWatchProgress (ctx , c , ep )
118126 if err != nil {
119- return false , err
127+ return err
120128 }
129+ f .lock .Lock ()
130+ defer f .lock .Unlock ()
121131
122- f .progresNotifyEndpointCache [ep ] = supported
123- return supported , nil
132+ if ! supported {
133+ klog .Infof ("RequestWatchProgress feature is not supported by %q endpoint" , ep )
134+ f .progressNotifySupported = ptr .To (false )
135+ return nil
136+ }
137+ if f .progressNotifySupported == nil {
138+ f .progressNotifySupported = ptr .To (true )
139+ }
140+ return nil
124141}
125142
126143// Sub interface of etcd client.
0 commit comments