mirror of
https://github.com/kubernetes/kubernetes.git
synced 2026-11-06 05:17:08 +00:00
metadatainformer and dynamic informer factory: add missing StartWithContext and WaitForCacheSyncWithContext
These methods already exist in informers.SharedInformerFactory. Because the factories here key informers by group/version/resource, WaitForCacheSyncWithContext returns the new cache.GVRSyncResult instead of cache.SyncResult. Start and WaitForCacheSync are now marked with //logcheck:context. Consumers which use the logcheck linter must switch to the *WithContext variants. Mock implementations of these interfaces need to be extended.
This commit is contained in:
1 parent
18789a4344
commit
5b1d3ee25e
11 files changed
+216
-58
No files matched your search
@@ -295,10 +295,14 @@ func Run(ctx context.Context, cc *schedulerserverconfig.CompletedConfig, sched *
|
||||
}
|
||||
|
||||
// Wait for all caches to sync before scheduling.
|
||||
cc.InformerFactory.WaitForCacheSyncWithContext(ctx)
|
||||
if err := cc.InformerFactory.WaitForCacheSyncWithContext(ctx).AsError(); err != nil {
|
||||
logger.Error(err, "informers are not fully synced")
|
||||
}
|
||||
// DynInformerFactory can be nil in tests.
|
||||
if cc.DynInformerFactory != nil {
|
||||
cc.DynInformerFactory.WaitForCacheSync(ctx.Done())
|
||||
if err := cc.DynInformerFactory.WaitForCacheSyncWithContext(ctx).AsError(); err != nil {
|
||||
logger.Error(err, "dynamic informers are not fully synced")
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for all handlers to sync (all items in the initial list delivered) before scheduling.
|
||||
|
||||
@@ -561,7 +561,7 @@ func TestAddAllEventHandlers(t *testing.T) {
|
||||
informerFactory.StartWithContext(ctx)
|
||||
dynInformerFactory.StartWithContext(ctx)
|
||||
staticInformers := informerFactory.WaitForCacheSyncWithContext(ctx).Synced
|
||||
dynamicInformers := dynInformerFactory.WaitForCacheSync(testSched.StopEverything)
|
||||
dynamicInformers := dynInformerFactory.WaitForCacheSyncWithContext(ctx).Synced
|
||||
|
||||
if diff := cmp.Diff(tt.expectStaticInformers, staticInformers); diff != "" {
|
||||
t.Errorf("Unexpected diff (-want, +got):\n%s", diff)
|
||||
@@ -619,7 +619,7 @@ func TestAddAllEventHandlersPodEventResources(t *testing.T) {
|
||||
informerFactory.StartWithContext(ctx)
|
||||
dynInformerFactory.StartWithContext(ctx)
|
||||
staticInformers := informerFactory.WaitForCacheSyncWithContext(ctx).Synced
|
||||
dynamicInformers := dynInformerFactory.WaitForCacheSync(testSched.StopEverything)
|
||||
dynamicInformers := dynInformerFactory.WaitForCacheSyncWithContext(ctx).Synced
|
||||
|
||||
expectStaticInformers := map[reflect.Type]bool{
|
||||
reflect.TypeFor[*v1.Pod](): true,
|
||||
|
||||
@@ -4,6 +4,23 @@ Go API changes are typically not included in the Kubernetes release notes, so
|
||||
noteworthy Go API changes *may* be documented here. This is currently not
|
||||
*required*, so consult the git history to see all changes.
|
||||
|
||||
### metadatainformer and dynamic informer factory: add missing StartWithContext and WaitForCacheSyncWithContext
|
||||
|
||||
These methods already exist in informers.SharedInformerFactory. Because the
|
||||
factories here key informers by group/version/resource,
|
||||
WaitForCacheSyncWithContext returns the new cache.GVRSyncResult instead of
|
||||
cache.SyncResult.
|
||||
|
||||
Start and WaitForCacheSync are now marked with //logcheck:context. Consumers
|
||||
which use the logcheck linter must switch to the *WithContext variants.
|
||||
Mock implementations of these interfaces need to be extended.
|
||||
|
||||
```
|
||||
- ./dynamic/dynamicinformer.DynamicSharedInformerFactory.WaitForCacheSyncWithContext: added
|
||||
- ./metadata/metadatainformer.SharedInformerFactory.StartWithContext: added
|
||||
- ./metadata/metadatainformer.SharedInformerFactory.WaitForCacheSyncWithContext: added
|
||||
```
|
||||
|
||||
### SharedInformer: add SetShardSelector
|
||||
|
||||
Mock implementations of the SharedInformer or SharedIndexInformer interfaces
|
||||
|
||||
@@ -154,7 +154,15 @@ func (f *dynamicSharedInformerFactory) StartWithContext(ctx context.Context) {
|
||||
}
|
||||
|
||||
// WaitForCacheSync waits for all started informers' cache were synced.
|
||||
//
|
||||
//logcheck:context // WaitForCacheSyncWithContext should be used instead of WaitForCacheSync in code which supports contextual logging.
|
||||
func (f *dynamicSharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) map[schema.GroupVersionResource]bool {
|
||||
return f.WaitForCacheSyncWithContext(wait.ContextForChannel(stopCh)).Synced
|
||||
}
|
||||
|
||||
// WaitForCacheSyncWithContext waits for all started informers' caches to be
|
||||
// synced or the context to be canceled.
|
||||
func (f *dynamicSharedInformerFactory) WaitForCacheSyncWithContext(ctx context.Context) cache.GVRSyncResult {
|
||||
informers := func() map[schema.GroupVersionResource]cache.SharedIndexInformer {
|
||||
f.lock.Lock()
|
||||
defer f.lock.Unlock()
|
||||
@@ -168,9 +176,27 @@ func (f *dynamicSharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{})
|
||||
return informers
|
||||
}()
|
||||
|
||||
res := map[schema.GroupVersionResource]bool{}
|
||||
// Wait for informers to sync, without polling.
|
||||
cacheSyncs := make([]cache.DoneChecker, 0, len(informers))
|
||||
for _, informer := range informers {
|
||||
cacheSyncs = append(cacheSyncs, informer.HasSyncedChecker())
|
||||
}
|
||||
cache.WaitFor(ctx, "" /* no logging */, cacheSyncs...)
|
||||
|
||||
res := cache.GVRSyncResult{
|
||||
Synced: make(map[schema.GroupVersionResource]bool, len(informers)),
|
||||
}
|
||||
failed := false
|
||||
for informType, informer := range informers {
|
||||
res[informType] = cache.WaitForCacheSync(stopCh, informer.HasSynced)
|
||||
hasSynced := informer.HasSynced()
|
||||
if !hasSynced {
|
||||
failed = true
|
||||
}
|
||||
res.Synced[informType] = hasSynced
|
||||
}
|
||||
if failed {
|
||||
// Non-nil, otherwise WaitFor wouldn't have stopped prematurely.
|
||||
res.Err = context.Cause(ctx)
|
||||
}
|
||||
return res
|
||||
}
|
||||
|
||||
@@ -251,8 +251,8 @@ func TestFilteredDynamicSharedInformerFactory(t *testing.T) {
|
||||
logger := klog.FromContext(ctx)
|
||||
_, _ = informerListerForGvr.Informer().AddEventHandlerWithOptions(ts.handler(informerReciveObjectCh), cache.HandlerOptions{Logger: &logger})
|
||||
target.StartWithContext(ctx)
|
||||
if synced := target.WaitForCacheSync(ctx.Done()); !synced[ts.gvr] {
|
||||
t.Errorf("informer for %s hasn't synced", ts.gvr)
|
||||
if result := target.WaitForCacheSyncWithContext(ctx); result.Err != nil || !result.Synced[ts.gvr] {
|
||||
t.Errorf("informer for %s hasn't synced: %v", ts.gvr, result.Err)
|
||||
}
|
||||
|
||||
testObject := ts.trigger(ts.gvr, ts.ns, fakeClient, ts.existingObj)
|
||||
@@ -376,8 +376,8 @@ func TestDynamicSharedInformerFactory(t *testing.T) {
|
||||
logger := klog.FromContext(ctx)
|
||||
_, _ = informerListerForGvr.Informer().AddEventHandlerWithOptions(ts.handler(informerReciveObjectCh), cache.HandlerOptions{Logger: &logger})
|
||||
target.StartWithContext(ctx)
|
||||
if synced := target.WaitForCacheSync(ctx.Done()); !synced[ts.gvr] {
|
||||
t.Errorf("informer for %s hasn't synced", ts.gvr)
|
||||
if result := target.WaitForCacheSyncWithContext(ctx); result.Err != nil || !result.Synced[ts.gvr] {
|
||||
t.Errorf("informer for %s hasn't synced: %v", ts.gvr, result.Err)
|
||||
}
|
||||
|
||||
testObject := ts.trigger(ts.gvr, ts.ns, fakeClient, ts.existingObj)
|
||||
|
||||
@@ -22,6 +22,7 @@ import (
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/client-go/informers"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
)
|
||||
|
||||
// DynamicSharedInformerFactory provides access to a shared informer and lister for dynamic client
|
||||
@@ -29,7 +30,7 @@ type DynamicSharedInformerFactory interface {
|
||||
// Start initializes all requested informers. They are handled in goroutines
|
||||
// which run until the stop channel gets closed.
|
||||
//
|
||||
// Contextual logging: StartWithContext should be used instead of Start in code which supports contextual logging.
|
||||
//logcheck:context // StartWithContext should be used instead of Start in code which supports contextual logging.
|
||||
Start(stopCh <-chan struct{})
|
||||
|
||||
// StartWithContext initializes all requested informers. They are handled in goroutines
|
||||
@@ -41,8 +42,14 @@ type DynamicSharedInformerFactory interface {
|
||||
|
||||
// WaitForCacheSync blocks until all started informers' caches were synced
|
||||
// or the stop channel gets closed.
|
||||
//
|
||||
//logcheck:context // WaitForCacheSyncWithContext should be used instead of WaitForCacheSync in code which supports contextual logging.
|
||||
WaitForCacheSync(stopCh <-chan struct{}) map[schema.GroupVersionResource]bool
|
||||
|
||||
// WaitForCacheSyncWithContext blocks until all started informers' caches were synced
|
||||
// or the context gets canceled.
|
||||
WaitForCacheSyncWithContext(ctx context.Context) cache.GVRSyncResult
|
||||
|
||||
// Shutdown marks a factory as shutting down. At that point no new
|
||||
// informers can be started anymore and Start will return without
|
||||
// doing anything.
|
||||
|
||||
@@ -164,7 +164,15 @@ func (f *metadataSharedInformerFactory) StartWithContext(ctx context.Context) {
|
||||
}
|
||||
|
||||
// WaitForCacheSync waits for all started informers' cache were synced.
|
||||
//
|
||||
//logcheck:context // WaitForCacheSyncWithContext should be used instead of WaitForCacheSync in code which supports contextual logging.
|
||||
func (f *metadataSharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) map[schema.GroupVersionResource]bool {
|
||||
return f.WaitForCacheSyncWithContext(wait.ContextForChannel(stopCh)).Synced
|
||||
}
|
||||
|
||||
// WaitForCacheSyncWithContext waits for all started informers' caches to be
|
||||
// synced or the context to be canceled.
|
||||
func (f *metadataSharedInformerFactory) WaitForCacheSyncWithContext(ctx context.Context) cache.GVRSyncResult {
|
||||
informers := func() map[schema.GroupVersionResource]cache.SharedIndexInformer {
|
||||
f.lock.Lock()
|
||||
defer f.lock.Unlock()
|
||||
@@ -178,9 +186,27 @@ func (f *metadataSharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{})
|
||||
return informers
|
||||
}()
|
||||
|
||||
res := map[schema.GroupVersionResource]bool{}
|
||||
// Wait for informers to sync, without polling.
|
||||
cacheSyncs := make([]cache.DoneChecker, 0, len(informers))
|
||||
for _, informer := range informers {
|
||||
cacheSyncs = append(cacheSyncs, informer.HasSyncedChecker())
|
||||
}
|
||||
cache.WaitFor(ctx, "" /* no logging */, cacheSyncs...)
|
||||
|
||||
res := cache.GVRSyncResult{
|
||||
Synced: make(map[schema.GroupVersionResource]bool, len(informers)),
|
||||
}
|
||||
failed := false
|
||||
for informType, informer := range informers {
|
||||
res[informType] = cache.WaitForCacheSync(stopCh, informer.HasSynced)
|
||||
hasSynced := informer.HasSynced()
|
||||
if !hasSynced {
|
||||
failed = true
|
||||
}
|
||||
res.Synced[informType] = hasSynced
|
||||
}
|
||||
if failed {
|
||||
// Non-nil, otherwise WaitFor wouldn't have stopped prematurely.
|
||||
res.Err = context.Cause(ctx)
|
||||
}
|
||||
return res
|
||||
}
|
||||
|
||||
@@ -18,7 +18,6 @@ package metadatainformer
|
||||
|
||||
import (
|
||||
"context"
|
||||
"flag"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
@@ -44,15 +43,9 @@ import (
|
||||
"k8s.io/client-go/metadata/fake"
|
||||
"k8s.io/client-go/rest"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
"k8s.io/klog/v2"
|
||||
"k8s.io/ktesting"
|
||||
)
|
||||
|
||||
func init() {
|
||||
klog.InitFlags(flag.CommandLine)
|
||||
flag.CommandLine.Lookup("v").Value.Set("5")
|
||||
flag.CommandLine.Lookup("alsologtostderr").Value.Set("true")
|
||||
}
|
||||
|
||||
func TestWatchListSemanticsSimple(t *testing.T) {
|
||||
clientfeaturestesting.SetFeatureDuringTest(t, clientfeatures.WatchListClient, true)
|
||||
|
||||
@@ -106,14 +99,13 @@ func TestWatchListSemanticsSimple(t *testing.T) {
|
||||
}
|
||||
|
||||
factory := NewFilteredSharedInformerFactory(client, 0, "ns", nil)
|
||||
target := factory.ForResource(schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"})
|
||||
_ = factory.ForResource(schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"})
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
factory.Start(ctx.Done())
|
||||
tCtx := ktesting.Init(t).WithTimeout(5*time.Second, "test timeout")
|
||||
factory.StartWithContext(tCtx)
|
||||
|
||||
if !cache.WaitForCacheSync(ctx.Done(), target.Informer().HasSynced) {
|
||||
t.Fatalf("failed to wait for caches to sync")
|
||||
if err := factory.WaitForCacheSyncWithContext(tCtx).AsError(); err != nil {
|
||||
t.Fatalf("failed to wait for caches to sync: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -127,14 +119,13 @@ func TestUnSupportWatchListSemantics(t *testing.T) {
|
||||
// so we don’t need to prepare a response.
|
||||
fakeClient := fake.NewSimpleMetadataClient(scheme)
|
||||
factory := NewFilteredSharedInformerFactory(fakeClient, 0, "ns", nil)
|
||||
target := factory.ForResource(schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"})
|
||||
_ = factory.ForResource(schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"})
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
factory.Start(ctx.Done())
|
||||
tCtx := ktesting.Init(t).WithTimeout(5*time.Second, "test timeout")
|
||||
factory.StartWithContext(tCtx)
|
||||
|
||||
if !cache.WaitForCacheSync(ctx.Done(), target.Informer().HasSynced) {
|
||||
t.Fatalf("failed to wait for caches to sync")
|
||||
if err := factory.WaitForCacheSyncWithContext(tCtx).AsError(); err != nil {
|
||||
t.Fatalf("failed to wait for caches to sync: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -144,7 +135,7 @@ func TestMetadataSharedInformerFactory(t *testing.T) {
|
||||
existingObj *metav1.PartialObjectMetadata
|
||||
gvr schema.GroupVersionResource
|
||||
ns string
|
||||
trigger func(gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeMetadataClient, testObject *metav1.PartialObjectMetadata) *metav1.PartialObjectMetadata
|
||||
trigger func(ctx context.Context, gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeMetadataClient, testObject *metav1.PartialObjectMetadata) *metav1.PartialObjectMetadata
|
||||
handler func(rcvCh chan<- *metav1.PartialObjectMetadata) *cache.ResourceEventHandlerFuncs
|
||||
}{
|
||||
// scenario 1
|
||||
@@ -152,7 +143,7 @@ func TestMetadataSharedInformerFactory(t *testing.T) {
|
||||
name: "scenario 1: test if adding an object triggers AddFunc",
|
||||
ns: "ns-foo",
|
||||
gvr: schema.GroupVersionResource{Group: "extensions", Version: "v1beta1", Resource: "deployments"},
|
||||
trigger: func(gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeMetadataClient, _ *metav1.PartialObjectMetadata) *metav1.PartialObjectMetadata {
|
||||
trigger: func(_ context.Context, gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeMetadataClient, _ *metav1.PartialObjectMetadata) *metav1.PartialObjectMetadata {
|
||||
testObject := newPartialObjectMetadata("extensions/v1beta1", "Deployment", "ns-foo", "name-foo")
|
||||
createdObj, err := fakeClient.Resource(gvr).Namespace(ns).(fake.MetadataClient).CreateFake(testObject, metav1.CreateOptions{})
|
||||
if err != nil {
|
||||
@@ -175,7 +166,7 @@ func TestMetadataSharedInformerFactory(t *testing.T) {
|
||||
ns: "ns-foo",
|
||||
gvr: schema.GroupVersionResource{Group: "extensions", Version: "v1beta1", Resource: "deployments"},
|
||||
existingObj: newPartialObjectMetadata("extensions/v1beta1", "Deployment", "ns-foo", "name-foo"),
|
||||
trigger: func(gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeMetadataClient, testObject *metav1.PartialObjectMetadata) *metav1.PartialObjectMetadata {
|
||||
trigger: func(_ context.Context, gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeMetadataClient, testObject *metav1.PartialObjectMetadata) *metav1.PartialObjectMetadata {
|
||||
if testObject.Annotations == nil {
|
||||
testObject.Annotations = make(map[string]string)
|
||||
}
|
||||
@@ -201,8 +192,8 @@ func TestMetadataSharedInformerFactory(t *testing.T) {
|
||||
ns: "ns-foo",
|
||||
gvr: schema.GroupVersionResource{Group: "extensions", Version: "v1beta1", Resource: "deployments"},
|
||||
existingObj: newPartialObjectMetadata("extensions/v1beta1", "Deployment", "ns-foo", "name-foo"),
|
||||
trigger: func(gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeMetadataClient, testObject *metav1.PartialObjectMetadata) *metav1.PartialObjectMetadata {
|
||||
err := fakeClient.Resource(gvr).Namespace(ns).Delete(context.TODO(), testObject.GetName(), metav1.DeleteOptions{})
|
||||
trigger: func(ctx context.Context, gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeMetadataClient, testObject *metav1.PartialObjectMetadata) *metav1.PartialObjectMetadata {
|
||||
err := fakeClient.Resource(gvr).Namespace(ns).Delete(ctx, testObject.GetName(), metav1.DeleteOptions{})
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
@@ -222,8 +213,7 @@ func TestMetadataSharedInformerFactory(t *testing.T) {
|
||||
t.Run(ts.name, func(t *testing.T) {
|
||||
// test data
|
||||
timeout := time.Duration(3 * time.Second)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
tCtx := ktesting.Init(t).WithTimeout(timeout, "test timeout")
|
||||
scheme := fake.NewTestScheme()
|
||||
metav1.AddMetaToScheme(scheme)
|
||||
informerReciveObjectCh := make(chan *metav1.PartialObjectMetadata, 1)
|
||||
@@ -236,20 +226,20 @@ func TestMetadataSharedInformerFactory(t *testing.T) {
|
||||
|
||||
// act
|
||||
informerListerForGvr := target.ForResource(ts.gvr)
|
||||
logger := klog.FromContext(ctx)
|
||||
logger := tCtx.Logger()
|
||||
_, _ = informerListerForGvr.Informer().AddEventHandlerWithOptions(ts.handler(informerReciveObjectCh), cache.HandlerOptions{Logger: &logger})
|
||||
target.Start(ctx.Done())
|
||||
if synced := target.WaitForCacheSync(ctx.Done()); !synced[ts.gvr] {
|
||||
t.Fatalf("informer for %s hasn't synced", ts.gvr)
|
||||
target.StartWithContext(tCtx)
|
||||
if result := target.WaitForCacheSyncWithContext(tCtx); result.Err != nil || !result.Synced[ts.gvr] {
|
||||
t.Fatalf("informer for %s hasn't synced: %v", ts.gvr, result.Err)
|
||||
}
|
||||
|
||||
testObject := ts.trigger(ts.gvr, ts.ns, fakeClient, ts.existingObj)
|
||||
testObject := ts.trigger(tCtx, ts.gvr, ts.ns, fakeClient, ts.existingObj)
|
||||
select {
|
||||
case objFromInformer := <-informerReciveObjectCh:
|
||||
if !equality.Semantic.DeepEqual(testObject, objFromInformer) {
|
||||
t.Fatalf("%v", cmp.Diff(testObject, objFromInformer))
|
||||
}
|
||||
case <-ctx.Done():
|
||||
case <-tCtx.Done():
|
||||
t.Errorf("tested informer haven't received an object, waited %v", timeout)
|
||||
}
|
||||
})
|
||||
@@ -270,8 +260,7 @@ func newPartialObjectMetadata(apiVersion, kind, namespace, name string) *metav1.
|
||||
}
|
||||
|
||||
func TestMetadataSharedInformerFactoryWithShardSelector(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
tCtx := ktesting.Init(t).WithTimeout(5*time.Second, "test timeout")
|
||||
|
||||
gvr := schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"}
|
||||
const initialObjects = 12
|
||||
@@ -303,10 +292,10 @@ func TestMetadataSharedInformerFactoryWithShardSelector(t *testing.T) {
|
||||
factories[shard] = NewSharedInformerFactoryWithOptions(fakeClient, 0, WithShardSelector(sel))
|
||||
informer := factories[shard].ForResource(gvr)
|
||||
_ = informer.Informer()
|
||||
factories[shard].Start(ctx.Done())
|
||||
factories[shard].StartWithContext(tCtx)
|
||||
t.Cleanup(factories[shard].Shutdown)
|
||||
if synced := factories[shard].WaitForCacheSync(ctx.Done()); !synced[gvr] {
|
||||
t.Fatalf("shard %d informer for %s failed to sync", shard, gvr)
|
||||
if result := factories[shard].WaitForCacheSyncWithContext(tCtx); result.Err != nil || !result.Synced[gvr] {
|
||||
t.Fatalf("shard %d informer for %s failed to sync: %v", shard, gvr, result.Err)
|
||||
}
|
||||
|
||||
items, err := informer.Lister().ByNamespace("ns-foo").List(labels.Everything())
|
||||
@@ -339,7 +328,7 @@ func TestMetadataSharedInformerFactoryWithShardSelector(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
if err := wait.PollUntilContextCancel(ctx, 10*time.Millisecond, true, func(_ context.Context) (bool, error) {
|
||||
if err := wait.PollUntilContextCancel(tCtx, 10*time.Millisecond, true, func(_ context.Context) (bool, error) {
|
||||
total := 0
|
||||
for shard := range totalShards {
|
||||
items, err := factories[shard].ForResource(gvr).Lister().ByNamespace("ns-foo").List(labels.Everything())
|
||||
|
||||
@@ -17,24 +17,39 @@ limitations under the License.
|
||||
package metadatainformer
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/client-go/informers"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
)
|
||||
|
||||
// SharedInformerFactory provides access to a shared informer and lister for dynamic client
|
||||
type SharedInformerFactory interface {
|
||||
// Start initializes all requested informers. They are handled in goroutines
|
||||
// which run until the stop channel gets closed.
|
||||
//
|
||||
//logcheck:context // StartWithContext should be used instead of Start in code which supports contextual logging.
|
||||
Start(stopCh <-chan struct{})
|
||||
|
||||
// StartWithContext initializes all requested informers. They are handled in goroutines
|
||||
// which run until the context gets canceled.
|
||||
StartWithContext(ctx context.Context)
|
||||
|
||||
// ForResource gives generic access to a shared informer of the matching type.
|
||||
ForResource(gvr schema.GroupVersionResource) informers.GenericInformer
|
||||
|
||||
// WaitForCacheSync blocks until all started informers' caches were synced
|
||||
// or the stop channel gets closed.
|
||||
//
|
||||
//logcheck:context // WaitForCacheSyncWithContext should be used instead of WaitForCacheSync in code which supports contextual logging.
|
||||
WaitForCacheSync(stopCh <-chan struct{}) map[schema.GroupVersionResource]bool
|
||||
|
||||
// WaitForCacheSyncWithContext blocks until all started informers' caches were synced
|
||||
// or the context gets canceled.
|
||||
WaitForCacheSyncWithContext(ctx context.Context) cache.GVRSyncResult
|
||||
|
||||
// Shutdown marks a factory as shutting down. At that point no new
|
||||
// informers can be started anymore and Start will return without
|
||||
// doing anything.
|
||||
|
||||
@@ -30,6 +30,7 @@ import (
|
||||
"k8s.io/apimachinery/pkg/api/meta"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/sharding"
|
||||
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
@@ -705,6 +706,37 @@ func (c SyncResult) AsError() error {
|
||||
return fmt.Errorf("failed to sync all caches: %s: %w", strings.Join(unsynced, ", "), c.Err)
|
||||
}
|
||||
|
||||
// GVRSyncResult is like [SyncResult] for factories which key their informers
|
||||
// by [schema.GroupVersionResource], i.e. the dynamic and metadata informer factories.
|
||||
type GVRSyncResult struct {
|
||||
// Err is nil if all informer caches were synced, otherwise it is
|
||||
// the reason why waiting for cache syncing stopped (= context.Cause(ctx)).
|
||||
Err error
|
||||
|
||||
// Synced maps each registered informer in a factory to
|
||||
// true if it has synced, false otherwise.
|
||||
Synced map[schema.GroupVersionResource]bool
|
||||
}
|
||||
|
||||
// AsError turns a GVRSyncResult into an error if not all caches were synced,
|
||||
// otherwise it returns nil. The error wraps context.Cause(ctx) and
|
||||
// includes information about the informers which were not synced.
|
||||
func (c GVRSyncResult) AsError() error {
|
||||
if c.Err == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
unsynced := make([]string, 0, len(c.Synced))
|
||||
for gvr, synced := range c.Synced {
|
||||
if !synced {
|
||||
unsynced = append(unsynced, gvr.String())
|
||||
}
|
||||
}
|
||||
slices.Sort(unsynced)
|
||||
|
||||
return fmt.Errorf("failed to sync all caches: %s: %w", strings.Join(unsynced, ", "), c.Err)
|
||||
}
|
||||
|
||||
// `*sharedIndexInformer` implements SharedIndexInformer and has three
|
||||
// main components. One is an indexed local cache, `indexer Indexer`.
|
||||
// The second main component is a Controller that pulls
|
||||
|
||||
+48
-6
@@ -30,6 +30,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/klog/v2"
|
||||
"k8s.io/klog/v2/textlogger"
|
||||
)
|
||||
@@ -201,14 +202,35 @@ func (m *mockChecker) Done() <-chan struct{} {
|
||||
}
|
||||
|
||||
func TestSyncResult(t *testing.T) {
|
||||
type syncResult interface {
|
||||
AsError() error
|
||||
}
|
||||
deployments := schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"}
|
||||
pods := schema.GroupVersionResource{Version: "v1", Resource: "pods"}
|
||||
cause := errors.New("my custom cancellation reason")
|
||||
|
||||
for name, tc := range map[string]struct {
|
||||
result SyncResult
|
||||
result syncResult
|
||||
expectAsError string
|
||||
}{
|
||||
"empty": {},
|
||||
"empty": {result: SyncResult{}},
|
||||
"gvr-empty": {result: GVRSyncResult{}},
|
||||
"all-synced": {
|
||||
result: SyncResult{
|
||||
Synced: map[reflect.Type]bool{
|
||||
reflect.TypeFor[int](): true,
|
||||
reflect.TypeFor[string](): true,
|
||||
},
|
||||
},
|
||||
},
|
||||
"gvr-all-synced": {
|
||||
result: GVRSyncResult{
|
||||
Synced: map[schema.GroupVersionResource]bool{deployments: true, pods: true},
|
||||
},
|
||||
},
|
||||
"one": {
|
||||
result: SyncResult{
|
||||
Err: errors.New("my custom cancellation reason"),
|
||||
Err: cause,
|
||||
Synced: map[reflect.Type]bool{
|
||||
reflect.TypeFor[int](): true,
|
||||
reflect.TypeFor[string](): false,
|
||||
@@ -216,9 +238,19 @@ func TestSyncResult(t *testing.T) {
|
||||
},
|
||||
expectAsError: "failed to sync all caches: string: my custom cancellation reason",
|
||||
},
|
||||
"gvr-one": {
|
||||
result: GVRSyncResult{
|
||||
Err: cause,
|
||||
Synced: map[schema.GroupVersionResource]bool{
|
||||
deployments: true,
|
||||
pods: false,
|
||||
},
|
||||
},
|
||||
expectAsError: "failed to sync all caches: " + pods.String() + ": my custom cancellation reason",
|
||||
},
|
||||
"many": {
|
||||
result: SyncResult{
|
||||
Err: errors.New("my custom cancellation reason"),
|
||||
Err: cause,
|
||||
Synced: map[reflect.Type]bool{
|
||||
reflect.TypeFor[int](): false,
|
||||
reflect.TypeFor[string](): false,
|
||||
@@ -226,6 +258,16 @@ func TestSyncResult(t *testing.T) {
|
||||
},
|
||||
expectAsError: "failed to sync all caches: int, string: my custom cancellation reason",
|
||||
},
|
||||
"gvr-many": {
|
||||
result: GVRSyncResult{
|
||||
Err: cause,
|
||||
Synced: map[schema.GroupVersionResource]bool{
|
||||
deployments: false,
|
||||
pods: false,
|
||||
},
|
||||
},
|
||||
expectAsError: "failed to sync all caches: " + pods.String() + ", " + deployments.String() + ": my custom cancellation reason",
|
||||
},
|
||||
} {
|
||||
|
||||
t.Run(name, func(t *testing.T) {
|
||||
@@ -238,8 +280,8 @@ func TestSyncResult(t *testing.T) {
|
||||
case tc.expectAsError != "" && actual != nil && actual.Error() != tc.expectAsError:
|
||||
t.Fatalf("expected %q, got %q", tc.expectAsError, actual.Error())
|
||||
}
|
||||
if tc.result.Err != nil && !errors.Is(actual, tc.result.Err) {
|
||||
t.Errorf("actual error %+v should wrap %v but doesn't", actual, tc.result.Err)
|
||||
if tc.expectAsError != "" && !errors.Is(actual, cause) {
|
||||
t.Errorf("actual error %+v should wrap %v but doesn't", actual, cause)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user