scottk/kubernetes
1
0
Fork 0
mirror of https://github.com/optim-enterprises-bv/kubernetes.git synced 2025-10-31 02:08:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #126387 from pohly/log-client-go-tools-cache-apis

Browse Source 
client-go/tools/cache: add APIs with context parameter
This commit is contained in:
Kubernetes Prow Robot
2024-12-18 20:00:09 +01:00
committed by GitHub
parent 5cfad4f858 4638ba9716
commit e305c33988
33 changed files with 1132 additions and 366 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
cmd/kube-controller-manager/app/controllermanager.go
View File

@@ -805,6 +805,7 @@ func startServiceAccountTokenController(ctx context.Context, controllerContext C
return nil, false, fmt.Errorf("failed to build token generator: %v", err)
}
tokenController, err := serviceaccountcontroller.NewTokensController(
logger,
controllerContext.InformerFactory.Core().V1().ServiceAccounts(),
controllerContext.InformerFactory.Core().V1().Secrets(),
rootClientBuilder.ClientOrDie("tokens-controller"),

1
hack/golangci-hints.yaml
View File

@@ -143,6 +143,7 @@ linters-settings: # please keep this alphabetized
contextual k8s.io/api/.*
contextual k8s.io/apimachinery/pkg/util/runtime/.*
contextual k8s.io/client-go/metadata/.*
contextual k8s.io/client-go/tools/cache/.*
contextual k8s.io/client-go/tools/events/.*
contextual k8s.io/client-go/tools/record/.*
contextual k8s.io/component-helpers/.*

1
hack/golangci-strict.yaml
View File

@@ -189,6 +189,7 @@ linters-settings: # please keep this alphabetized
contextual k8s.io/api/.*
contextual k8s.io/apimachinery/pkg/util/runtime/.*
contextual k8s.io/client-go/metadata/.*
contextual k8s.io/client-go/tools/cache/.*
contextual k8s.io/client-go/tools/events/.*
contextual k8s.io/client-go/tools/record/.*
contextual k8s.io/component-helpers/.*

1
hack/golangci.yaml
View File

@@ -191,6 +191,7 @@ linters-settings: # please keep this alphabetized
contextual k8s.io/api/.*
contextual k8s.io/apimachinery/pkg/util/runtime/.*
contextual k8s.io/client-go/metadata/.*
contextual k8s.io/client-go/tools/cache/.*
contextual k8s.io/client-go/tools/events/.*
contextual k8s.io/client-go/tools/record/.*
contextual k8s.io/component-helpers/.*

1
hack/logcheck.conf
View File

@@ -27,6 +27,7 @@ structured k8s.io/apiserver/pkg/server/options/encryptionconfig/.*
contextual k8s.io/api/.*
contextual k8s.io/apimachinery/pkg/util/runtime/.*
contextual k8s.io/client-go/metadata/.*
contextual k8s.io/client-go/tools/cache/.*
contextual k8s.io/client-go/tools/events/.*
contextual k8s.io/client-go/tools/record/.*
contextual k8s.io/component-helpers/.*

10
pkg/controller/resourceclaim/controller.go
View File

@@ -146,7 +146,7 @@ func NewController(
metrics.RegisterMetrics()
if _, err := podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
if _, err := podInformer.Informer().AddEventHandlerWithOptions(cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
ec.enqueuePod(logger, obj, false)
},
@@ -156,10 +156,10 @@ func NewController(
DeleteFunc: func(obj interface{}) {
ec.enqueuePod(logger, obj, true)
},
}); err != nil {
}, cache.HandlerOptions{Logger: &logger}); err != nil {
return nil, err
}
if _, err := claimInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
if _, err := claimInformer.Informer().AddEventHandlerWithOptions(cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
logger.V(6).Info("new claim", "claimDump", obj)
ec.enqueueResourceClaim(logger, nil, obj)
@@ -172,7 +172,7 @@ func NewController(
logger.V(6).Info("deleted claim", "claimDump", obj)
ec.enqueueResourceClaim(logger, obj, nil)
},
}); err != nil {
}, cache.HandlerOptions{Logger: &logger}); err != nil {
return nil, err
}
if err := ec.podIndexer.AddIndexers(cache.Indexers{podResourceClaimIndex: podResourceClaimIndexFunc}); err != nil {
@@ -190,7 +190,7 @@ func NewController(
if err := claimInformerCache.AddIndexers(cache.Indexers{claimPodOwnerIndex: claimPodOwnerIndexFunc}); err != nil {
return nil, fmt.Errorf("could not initialize ResourceClaim controller: %w", err)
}
ec.claimCache = cache.NewIntegerResourceVersionMutationCache(claimInformerCache, claimInformerCache,
ec.claimCache = cache.NewIntegerResourceVersionMutationCache(logger, claimInformerCache, claimInformerCache,
// Very long time to live, unlikely to be needed because
// the informer cache should get updated soon.
time.Hour,

11
pkg/controller/serviceaccount/tokens_controller.go
View File

@@ -69,7 +69,7 @@ type TokensControllerOptions struct {
}
// NewTokensController returns a new *TokensController.
func NewTokensController(serviceAccounts informers.ServiceAccountInformer, secrets informers.SecretInformer, cl clientset.Interface, options TokensControllerOptions) (*TokensController, error) {
func NewTokensController(logger klog.Logger, serviceAccounts informers.ServiceAccountInformer, secrets informers.SecretInformer, cl clientset.Interface, options TokensControllerOptions) (*TokensController, error) {
maxRetries := options.MaxRetries
if maxRetries == 0 {
maxRetries = 10
@@ -104,9 +104,9 @@ func NewTokensController(serviceAccounts informers.ServiceAccountInformer, secre
)
secretCache := secrets.Informer().GetIndexer()
e.updatedSecrets = cache.NewIntegerResourceVersionMutationCache(secretCache, secretCache, 60*time.Second, true)
e.updatedSecrets = cache.NewIntegerResourceVersionMutationCache(logger, secretCache, secretCache, 60*time.Second, true)
e.secretSynced = secrets.Informer().HasSynced
secrets.Informer().AddEventHandlerWithResyncPeriod(
secrets.Informer().AddEventHandlerWithOptions(
cache.FilteringResourceEventHandler{
FilterFunc: func(obj interface{}) bool {
switch t := obj.(type) {
@@ -123,7 +123,10 @@ func NewTokensController(serviceAccounts informers.ServiceAccountInformer, secre
DeleteFunc: e.queueSecretSync,
},
},
options.SecretResync,
cache.HandlerOptions{
Logger: &logger,
ResyncPeriod: &options.SecretResync,
},
)
return e, nil

4
pkg/controller/serviceaccount/tokens_controller_test.go
View File

@@ -440,7 +440,7 @@ func TestTokenCreation(t *testing.T) {
for k, tc := range testcases {
t.Run(k, func(t *testing.T) {
_, ctx := ktesting.NewTestContext(t)
logger, ctx := ktesting.NewTestContext(t)
// Re-seed to reset name generation
utilrand.Seed(1)
@@ -455,7 +455,7 @@ func TestTokenCreation(t *testing.T) {
secretInformer := informers.Core().V1().Secrets().Informer()
secrets := secretInformer.GetStore()
serviceAccounts := informers.Core().V1().ServiceAccounts().Informer().GetStore()
controller, err := NewTokensController(informers.Core().V1().ServiceAccounts(), informers.Core().V1().Secrets(), client, TokensControllerOptions{TokenGenerator: generator, RootCA: []byte("CA Data"), MaxRetries: tc.MaxRetries})
controller, err := NewTokensController(logger, informers.Core().V1().ServiceAccounts(), informers.Core().V1().Secrets(), client, TokensControllerOptions{TokenGenerator: generator, RootCA: []byte("CA Data"), MaxRetries: tc.MaxRetries})
if err != nil {
t.Fatalf("error creating Tokens controller: %v", err)
}

3
staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/apiserver.go
View File

@@ -48,6 +48,7 @@ import (
genericapiserver "k8s.io/apiserver/pkg/server"
serverstorage "k8s.io/apiserver/pkg/server/storage"
"k8s.io/apiserver/pkg/util/webhook"
"k8s.io/klog/v2"
)
var (
@@ -211,7 +212,7 @@ func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget)
aggregatedDiscoveryManager = aggregatedDiscoveryManager.WithSource(aggregated.CRDSource)
}
discoveryController := NewDiscoveryController(s.Informers.Apiextensions().V1().CustomResourceDefinitions(), versionDiscoveryHandler, groupDiscoveryHandler, aggregatedDiscoveryManager)
namingController := status.NewNamingConditionController(s.Informers.Apiextensions().V1().CustomResourceDefinitions(), crdClient.ApiextensionsV1())
namingController := status.NewNamingConditionController(klog.TODO() /* for contextual logging */, s.Informers.Apiextensions().V1().CustomResourceDefinitions(), crdClient.ApiextensionsV1())
nonStructuralSchemaController := nonstructuralschema.NewConditionController(s.Informers.Apiextensions().V1().CustomResourceDefinitions(), crdClient.ApiextensionsV1())
apiApprovalController := apiapproval.NewKubernetesAPIApprovalPolicyConformantConditionController(s.Informers.Apiextensions().V1().CustomResourceDefinitions(), crdClient.ApiextensionsV1())
finalizingController := finalizer.NewCRDFinalizer(

11
staging/src/k8s.io/apiextensions-apiserver/pkg/controller/status/naming_controller.go
View File

@@ -62,6 +62,7 @@ type NamingConditionController struct {
}
func NewNamingConditionController(
logger klog.Logger,
crdInformer informers.CustomResourceDefinitionInformer,
crdClient client.CustomResourceDefinitionsGetter,
) *NamingConditionController {
@@ -76,13 +77,17 @@ func NewNamingConditionController(
}
informerIndexer := crdInformer.Informer().GetIndexer()
c.crdMutationCache = cache.NewIntegerResourceVersionMutationCache(informerIndexer, informerIndexer, 60*time.Second, false)
c.crdMutationCache = cache.NewIntegerResourceVersionMutationCache(logger, informerIndexer, informerIndexer, 60*time.Second, false)
crdInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
crdInformer.Informer().AddEventHandlerWithOptions(cache.ResourceEventHandlerFuncs{
AddFunc: c.addCustomResourceDefinition,
UpdateFunc: c.updateCustomResourceDefinition,
DeleteFunc: c.deleteCustomResourceDefinition,
})
},
cache.HandlerOptions{
Logger: &logger,
},
)
c.syncFn = c.sync

16
staging/src/k8s.io/apiextensions-apiserver/pkg/controller/status/naming_controller_test.go
View File

@@ -27,6 +27,7 @@ import (
listers "k8s.io/apiextensions-apiserver/pkg/client/listers/apiextensions/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/tools/cache"
"k8s.io/klog/v2/ktesting"
)
type crdBuilder struct {
@@ -321,25 +322,30 @@ func TestSync(t *testing.T) {
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
logger, _ := ktesting.NewTestContext(t)
crdIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
for _, obj := range tc.existing {
crdIndexer.Add(obj)
if err := crdIndexer.Add(obj); err != nil {
t.Fatalf("unexpected error: %v", err)
}
}
c := NamingConditionController{
crdLister: listers.NewCustomResourceDefinitionLister(crdIndexer),
crdMutationCache: cache.NewIntegerResourceVersionMutationCache(crdIndexer, crdIndexer, 60*time.Second, false),
crdMutationCache: cache.NewIntegerResourceVersionMutationCache(logger, crdIndexer, crdIndexer, 60*time.Second, false),
}
actualNames, actualNameConflictCondition, establishedCondition := c.calculateNamesAndConditions(tc.in)
if e, a := tc.expectedNames, actualNames; !reflect.DeepEqual(e, a) {
t.Errorf("%v expected %v, got %#v", tc.name, e, a)
t.Errorf("expected %v, got %#v", e, a)
}
if e, a := tc.expectedNameConflictCondition, actualNameConflictCondition; !apiextensionshelpers.IsCRDConditionEquivalent(&e, &a) {
t.Errorf("%v expected %v, got %v", tc.name, e, a)
t.Errorf("expected %v, got %v", e, a)
}
if e, a := tc.expectedEstablishedCondition, establishedCondition; !apiextensionshelpers.IsCRDConditionEquivalent(&e, &a) {
t.Errorf("%v expected %v, got %v", tc.name, e, a)
t.Errorf("expected %v, got %v", e, a)
}
})
}
}

46
staging/src/k8s.io/apimachinery/pkg/util/runtime/runtime.go
View File

@@ -36,6 +36,11 @@ var (
)
// PanicHandlers is a list of functions which will be invoked when a panic happens.
//
// The code invoking these handlers prepares a contextual logger so that
// klog.FromContext(ctx) already skips over the panic handler itself and
// several other intermediate functions, ideally such that the log output
// is attributed to the code which triggered the panic.
var PanicHandlers = []func(context.Context, interface{}){logPanic}
// HandleCrash simply catches a crash and logs an error. Meant to be called via
@@ -45,7 +50,7 @@ var PanicHandlers = []func(context.Context, interface{}){logPanic}
//
// E.g., you can provide one or more additional handlers for something like shutting down go routines gracefully.
//
// Contextual logging: HandleCrashWithContext should be used instead of HandleCrash in code which supports contextual logging.
// Contextual logging: HandleCrashWithContext or HandleCrashWithLogger should be used instead of HandleCrash in code which supports contextual logging.
func HandleCrash(additionalHandlers ...func(interface{})) {
if r := recover(); r != nil {
additionalHandlersWithContext := make([]func(context.Context, interface{}), len(additionalHandlers))
@@ -74,10 +79,30 @@ func HandleCrashWithContext(ctx context.Context, additionalHandlers ...func(cont
}
}
// handleCrash is the common implementation of HandleCrash and HandleCrash.
// HandleCrashWithLogger simply catches a crash and logs an error. Meant to be called via
// defer. Additional context-specific handlers can be provided, and will be
// called in case of panic. HandleCrash actually crashes, after calling the
// handlers and logging the panic message.
//
// E.g., you can provide one or more additional handlers for something like shutting down go routines gracefully.
func HandleCrashWithLogger(logger klog.Logger, additionalHandlers ...func(context.Context, interface{})) {
if r := recover(); r != nil {
ctx := klog.NewContext(context.Background(), logger)
handleCrash(ctx, r, additionalHandlers...)
}
}
// handleCrash is the common implementation of the HandleCrash* variants.
// Having those call a common implementation ensures that the stack depth
// is the same regardless through which path the handlers get invoked.
func handleCrash(ctx context.Context, r any, additionalHandlers ...func(context.Context, interface{})) {
// We don't really know how many call frames to skip because the Go
// panic handler is between us and the code where the panic occurred.
// If it's one function (as in Go 1.21), then skipping four levels
// gets us to the function which called the `defer HandleCrashWithontext(...)`.
logger := klog.FromContext(ctx).WithCallDepth(4)
ctx = klog.NewContext(ctx, logger)
for _, fn := range PanicHandlers {
fn(ctx, r)
}
@@ -106,11 +131,7 @@ func logPanic(ctx context.Context, r interface{}) {
stacktrace := make([]byte, size)
stacktrace = stacktrace[:runtime.Stack(stacktrace, false)]
// We don't really know how many call frames to skip because the Go
// panic handler is between us and the code where the panic occurred.
// If it's one function (as in Go 1.21), then skipping four levels
// gets us to the function which called the `defer HandleCrashWithontext(...)`.
logger := klog.FromContext(ctx).WithCallDepth(4)
logger := klog.FromContext(ctx)
// For backwards compatibility, conversion to string
// is handled here instead of defering to the logging
@@ -176,12 +197,19 @@ func HandleError(err error) {
// and key/value pairs.
//
// This variant should be used instead of HandleError because it supports
// structured, contextual logging.
// structured, contextual logging. Alternatively, [HandleErrorWithLogger] can
// be used if a logger is available instead of a context.
func HandleErrorWithContext(ctx context.Context, err error, msg string, keysAndValues ...interface{}) {
handleError(ctx, err, msg, keysAndValues...)
}
// handleError is the common implementation of HandleError and HandleErrorWithContext.
// HandleErrorWithLogger is an alternative to [HandlerErrorWithContext] which accepts
// a logger for contextual logging.
func HandleErrorWithLogger(logger klog.Logger, err error, msg string, keysAndValues ...interface{}) {
handleError(klog.NewContext(context.Background(), logger), err, msg, keysAndValues...)
}
// handleError is the common implementation of the HandleError* variants.
// Using this common implementation ensures that the stack depth
// is the same regardless through which path the handlers get invoked.
func handleError(ctx context.Context, err error, msg string, keysAndValues ...interface{}) {

127
staging/src/k8s.io/apimachinery/pkg/util/runtime/runtime_test.go
View File

@@ -19,15 +19,22 @@ package runtime
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"net/http"
"os"
"regexp"
"runtime"
"strings"
"sync"
"testing"
"time"
"github.com/stretchr/testify/assert"
"k8s.io/klog/v2"
"k8s.io/klog/v2/textlogger"
)
func TestHandleCrash(t *testing.T) {
@@ -36,6 +43,7 @@ func TestHandleCrash(t *testing.T) {
t.Errorf("Expected a panic to recover from")
}
}()
//nolint:logcheck // Intentionally uses the old API.
defer HandleCrash()
panic("Test Panic")
}
@@ -55,6 +63,7 @@ func TestCustomHandleCrash(t *testing.T) {
t.Errorf("Expected a panic to recover from")
}
}()
//nolint:logcheck // Intentionally uses the old API.
defer HandleCrash()
panic("test")
}()
@@ -73,6 +82,7 @@ func TestCustomHandleError(t *testing.T) {
},
}
err := fmt.Errorf("test")
//nolint:logcheck // Intentionally uses the old API.
HandleError(err)
if result != err {
t.Errorf("did not receive custom handler")
@@ -86,6 +96,7 @@ func TestHandleCrashLog(t *testing.T) {
t.Fatalf("expected a panic to recover from")
}
}()
//nolint:logcheck // Intentionally uses the old API.
defer HandleCrash()
panic("test panic")
})
@@ -119,6 +130,72 @@ func TestHandleCrashLog(t *testing.T) {
}
}
func TestHandleCrashContextual(t *testing.T) {
for name, handleCrash := range map[string]func(logger klog.Logger, trigger func(), additionalHandlers ...func(context.Context, interface{})){
"WithLogger": func(logger klog.Logger, trigger func(), additionalHandlers ...func(context.Context, interface{})) {
logger = logger.WithCallDepth(2) // This function *and* the trigger helper.
defer HandleCrashWithLogger(logger, additionalHandlers...)
trigger()
},
"WithContext": func(logger klog.Logger, trigger func(), additionalHandlers ...func(context.Context, interface{})) {
logger = logger.WithCallDepth(2)
defer HandleCrashWithContext(klog.NewContext(context.Background(), logger), additionalHandlers...)
trigger()
},
} {
t.Run(name, func(t *testing.T) {
for name, tt := range map[string]struct {
trigger func()
expectPanic string
}{
"no-panic": {
trigger: func() {},
expectPanic: "",
},
"string-panic": {
trigger: func() { panic("fake") },
expectPanic: "fake",
},
"int-panic": {
trigger: func() { panic(42) },
expectPanic: "42",
},
} {
t.Run(name, func(t *testing.T) {
var buffer bytes.Buffer
timeInUTC := time.Date(2009, 12, 1, 13, 30, 40, 42000, time.UTC)
timeString := "1201 13:30:40.000042"
logger := textlogger.NewLogger(textlogger.NewConfig(
textlogger.FixedTime(timeInUTC),
textlogger.Output(&buffer),
))
ReallyCrash = false
defer func() { ReallyCrash = true }()
handler := func(ctx context.Context, r interface{}) {
// Same formatting as in HandleCrash.
str, ok := r.(string)
if !ok {
str = fmt.Sprintf("%v", r)
}
klog.FromContext(ctx).Info("handler called", "panic", str)
}
_, _, line, _ := runtime.Caller(0)
handleCrash(logger, tt.trigger, handler)
if tt.expectPanic != "" {
assert.Contains(t, buffer.String(), fmt.Sprintf(`E%s %7d runtime_test.go:%d] "Observed a panic" panic=%q`, timeString, os.Getpid(), line+1, tt.expectPanic))
assert.Contains(t, buffer.String(), fmt.Sprintf(`I%s %7d runtime_test.go:%d] "handler called" panic=%q
`, timeString, os.Getpid(), line+1, tt.expectPanic))
} else {
assert.Empty(t, buffer.String())
}
})
}
})
}
}
func TestHandleCrashLogSilenceHTTPErrAbortHandler(t *testing.T) {
log, err := captureStderr(func() {
defer func() {
@@ -126,6 +203,7 @@ func TestHandleCrashLogSilenceHTTPErrAbortHandler(t *testing.T) {
t.Fatalf("expected to recover from http.ErrAbortHandler")
}
}()
//nolint:logcheck // Intentionally uses the old API.
defer HandleCrash()
panic(http.ErrAbortHandler)
})
@@ -184,3 +262,52 @@ func Test_rudimentaryErrorBackoff_OnError_ParallelSleep(t *testing.T) {
t.Errorf("OnError slept for too long: %s", since)
}
}
func TestHandleError(t *testing.T) {
for name, handleError := range map[string]func(logger klog.Logger, err error, msg string, keysAndValues ...interface{}){
"WithLogger": func(logger klog.Logger, err error, msg string, keysAndValues ...interface{}) {
helper, logger := logger.WithCallStackHelper()
helper()
HandleErrorWithLogger(logger, err, msg, keysAndValues...)
},
"WithContext": func(logger klog.Logger, err error, msg string, keysAndValues ...interface{}) {
helper, logger := logger.WithCallStackHelper()
helper()
HandleErrorWithContext(klog.NewContext(context.Background(), logger), err, msg, keysAndValues...)
},
} {
t.Run(name, func(t *testing.T) {
for name, tc := range map[string]struct {
err error
msg string
keysAndValues []interface{}
expectLog string
}{
"no-error": {
msg: "hello world",
expectLog: `"hello world" logger="UnhandledError"`,
},
"complex": {
err: errors.New("fake error"),
msg: "ignore",
keysAndValues: []interface{}{"a", 1, "b", "c"},
expectLog: `"ignore" err="fake error" logger="UnhandledError" a=1 b="c"`,
},
} {
t.Run(name, func(t *testing.T) {
var buffer bytes.Buffer
timeInUTC := time.Date(2009, 12, 1, 13, 30, 40, 42000, time.UTC)
timeString := "1201 13:30:40.000042"
logger := textlogger.NewLogger(textlogger.NewConfig(
textlogger.FixedTime(timeInUTC),
textlogger.Output(&buffer),
))
_, _, line, _ := runtime.Caller(0)
handleError(logger, tc.err, tc.msg, tc.keysAndValues...)
assert.Equal(t, fmt.Sprintf("E%s %7d runtime_test.go:%d] %s\n", timeString, os.Getpid(), line+1, tc.expectLog), buffer.String())
})
}
})
}
}

24
staging/src/k8s.io/client-go/examples/workqueue/main.go
View File

@@ -17,6 +17,8 @@ limitations under the License.
package main
import (
"context"
"errors"
"flag"
"fmt"
"time"
@@ -116,30 +118,30 @@ func (c *Controller) handleErr(err error, key string) {
}
// Run begins watching and syncing.
func (c *Controller) Run(workers int, stopCh chan struct{}) {
defer runtime.HandleCrash()
func (c *Controller) Run(ctx context.Context, workers int) {
defer runtime.HandleCrashWithContext(ctx)
// Let the workers stop when we are done
defer c.queue.ShutDown()
klog.Info("Starting Pod controller")
go c.informer.Run(stopCh)
go c.informer.RunWithContext(ctx)
// Wait for all involved caches to be synced, before processing items from the queue is started
if !cache.WaitForCacheSync(stopCh, c.informer.HasSynced) {
if !cache.WaitForNamedCacheSyncWithContext(ctx, c.informer.HasSynced) {
runtime.HandleError(fmt.Errorf("Timed out waiting for caches to sync"))
return
}
for i := 0; i < workers; i++ {
go wait.Until(c.runWorker, time.Second, stopCh)
go wait.UntilWithContext(ctx, c.runWorker, time.Second)
}
<-stopCh
<-ctx.Done()
klog.Info("Stopping Pod controller")
}
func (c *Controller) runWorker() {
func (c *Controller) runWorker(ctx context.Context) {
for c.processNextItem() {
}
}
@@ -164,6 +166,8 @@ func main() {
klog.Fatal(err)
}
ctx := context.Background()
// create the pod watcher
podListWatcher := cache.NewListWatchFromClient(clientset.CoreV1().RESTClient(), "pods", v1.NamespaceDefault, fields.Everything())
@@ -211,9 +215,9 @@ func main() {
})
// Now let's start the controller
stop := make(chan struct{})
defer close(stop)
go controller.Run(1, stop)
cancelCtx, cancel := context.WithCancelCause(ctx)
defer cancel(errors.New("time to stop because main has completed"))
go controller.Run(cancelCtx, 1)
// Wait forever
select {}

25
staging/src/k8s.io/client-go/tools/cache/cache_test.go vendored Normal file
View File

@@ -0,0 +1,25 @@
/*
Copyright 2024 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package cache
import (
"k8s.io/klog/v2"
)
func init() {
klog.InitFlags(nil)
}

53
staging/src/k8s.io/client-go/tools/cache/controller.go vendored
View File

@@ -17,6 +17,7 @@ limitations under the License.
package cache
import (
"context"
"errors"
"sync"
"time"
@@ -79,8 +80,14 @@ type Config struct {
RetryOnError bool
// Called whenever the ListAndWatch drops the connection with an error.
//
// Contextual logging: WatchErrorHandlerWithContext should be used instead of WatchErrorHandler in code which supports contextual logging.
WatchErrorHandler WatchErrorHandler
// Called whenever the ListAndWatch drops the connection with an error
// and WatchErrorHandler is not set.
WatchErrorHandlerWithContext WatchErrorHandlerWithContext
// WatchListPageSize is the requested chunk size of initial and relist watch lists.
WatchListPageSize int64
}
@@ -104,12 +111,21 @@ type controller struct {
// Controller is a low-level controller that is parameterized by a
// Config and used in sharedIndexInformer.
type Controller interface {
// Run does two things. One is to construct and run a Reflector
// RunWithContext does two things. One is to construct and run a Reflector
// to pump objects/notifications from the Config's ListerWatcher
// to the Config's Queue and possibly invoke the occasional Resync
// on that Queue. The other is to repeatedly Pop from the Queue
// and process with the Config's ProcessFunc. Both of these
// continue until `stopCh` is closed.
// continue until the context is canceled.
//
// It's an error to call RunWithContext more than once.
// RunWithContext blocks; call via go.
RunWithContext(ctx context.Context)
// Run does the same as RunWithContext with a stop channel instead of
// a context.
//
// Contextual logging: RunWithcontext should be used instead of Run in code which supports contextual logging.
Run(stopCh <-chan struct{})
// HasSynced delegates to the Config's Queue
@@ -129,13 +145,16 @@ func New(c *Config) Controller {
return ctlr
}
// Run begins processing items, and will continue until a value is sent down stopCh or it is closed.
// It's an error to call Run more than once.
// Run blocks; call via go.
// Run implements [Controller.Run].
func (c *controller) Run(stopCh <-chan struct{}) {
defer utilruntime.HandleCrash()
c.RunWithContext(wait.ContextForChannel(stopCh))
}
// RunWithContext implements [Controller.RunWithContext].
func (c *controller) RunWithContext(ctx context.Context) {
defer utilruntime.HandleCrashWithContext(ctx)
go func() {
<-stopCh
<-ctx.Done()
c.config.Queue.Close()
}()
r := NewReflectorWithOptions(
@@ -152,7 +171,11 @@ func (c *controller) Run(stopCh <-chan struct{}) {
r.ShouldResync = c.config.ShouldResync
r.WatchListPageSize = c.config.WatchListPageSize
if c.config.WatchErrorHandler != nil {
r.watchErrorHandler = c.config.WatchErrorHandler
r.watchErrorHandler = func(_ context.Context, r *Reflector, err error) {
c.config.WatchErrorHandler(r, err)
}
} else if c.config.WatchErrorHandlerWithContext != nil {
r.watchErrorHandler = c.config.WatchErrorHandlerWithContext
}
c.reflectorMutex.Lock()
@@ -161,9 +184,9 @@ func (c *controller) Run(stopCh <-chan struct{}) {
var wg wait.Group
wg.StartWithChannel(stopCh, r.Run)
wg.StartWithContext(ctx, r.RunWithContext)
wait.Until(c.processLoop, time.Second, stopCh)
wait.UntilWithContext(ctx, c.processLoop, time.Second)
wg.Wait()
}
@@ -185,13 +208,11 @@ func (c *controller) LastSyncResourceVersion() string {
// TODO: Consider doing the processing in parallel. This will require a little thought
// to make sure that we don't end up processing the same object multiple times
// concurrently.
//
// TODO: Plumb through the stopCh here (and down to the queue) so that this can
// actually exit when the controller is stopped. Or just give up on this stuff
// ever being stoppable. Converting this whole package to use Context would
// also be helpful.
func (c *controller) processLoop() {
func (c *controller) processLoop(ctx context.Context) {
for {
// TODO: Plumb through the ctx so that this can
// actually exit when the controller is stopped. Or just give up on this stuff
// ever being stoppable.
obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
if err != nil {
if err == ErrFIFOClosed {

86
staging/src/k8s.io/client-go/tools/cache/controller_test.go vendored
View File

@@ -17,6 +17,7 @@ limitations under the License.
package cache
import (
"context"
"fmt"
"math/rand"
"sync"
@@ -32,6 +33,7 @@ import (
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/apimachinery/pkg/watch"
fcache "k8s.io/client-go/tools/cache/testing"
"k8s.io/klog/v2/ktesting"
fuzz "github.com/google/gofuzz"
)
@@ -39,6 +41,7 @@ import (
func Example() {
// source simulates an apiserver object endpoint.
source := fcache.NewFakeControllerSource()
defer source.Shutdown()
// This will hold the downstream state, as we know it.
downstream := NewStore(DeletionHandlingMetaNamespaceKeyFunc)
@@ -97,10 +100,10 @@ func Example() {
},
}
// Create the controller and run it until we close stop.
stop := make(chan struct{})
defer close(stop)
go New(cfg).Run(stop)
// Create the controller and run it until we cancel.
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go New(cfg).RunWithContext(ctx)
// Let's add a few objects to the source.
testIDs := []string{"a-hello", "b-controller", "c-framework"}
@@ -128,6 +131,7 @@ func Example() {
func ExampleNewInformer() {
// source simulates an apiserver object endpoint.
source := fcache.NewFakeControllerSource()
defer source.Shutdown()
// Let's do threadsafe output to get predictable test results.
deletionCounter := make(chan string, 1000)
@@ -154,10 +158,10 @@ func ExampleNewInformer() {
},
)
// Run the controller and run it until we close stop.
stop := make(chan struct{})
defer close(stop)
go controller.Run(stop)
// Run the controller and run it until we cancel.
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go controller.RunWithContext(ctx)
// Let's add a few objects to the source.
testIDs := []string{"a-hello", "b-controller", "c-framework"}
@@ -189,7 +193,7 @@ func TestHammerController(t *testing.T) {
// race detector.
// source simulates an apiserver object endpoint.
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
// Let's do threadsafe output to get predictable test results.
outputSetLock := sync.Mutex{}
@@ -225,9 +229,10 @@ func TestHammerController(t *testing.T) {
t.Errorf("Expected HasSynced() to return false before we started the controller")
}
// Run the controller and run it until we close stop.
stop := make(chan struct{})
go controller.Run(stop)
// Run the controller and run it until we cancel.
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancel(ctx)
go controller.RunWithContext(ctx)
// Let's wait for the controller to do its initial sync
wait.Poll(100*time.Millisecond, wait.ForeverTestTimeout, func() (bool, error) {
@@ -286,7 +291,7 @@ func TestHammerController(t *testing.T) {
// Let's wait for the controller to finish processing the things we just added.
// TODO: look in the queue to see how many items need to be processed.
time.Sleep(100 * time.Millisecond)
close(stop)
cancel()
// TODO: Verify that no goroutines were leaked here and that everything shut
// down cleanly.
@@ -300,7 +305,7 @@ func TestUpdate(t *testing.T) {
// call to update.
// source simulates an apiserver object endpoint.
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
const (
FROM = "from"
@@ -383,11 +388,13 @@ func TestUpdate(t *testing.T) {
},
)
// Run the controller and run it until we close stop.
// Run the controller and run it until we cancel.
// Once Run() is called, calls to testDoneWG.Done() might start, so
// all testDoneWG.Add() calls must happen before this point
stop := make(chan struct{})
go controller.Run(stop)
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancel(ctx)
defer cancel()
go controller.RunWithContext(ctx)
<-watchCh
// run every test a few times, in parallel
@@ -405,12 +412,11 @@ func TestUpdate(t *testing.T) {
// Let's wait for the controller to process the things we just added.
testDoneWG.Wait()
close(stop)
}
func TestPanicPropagated(t *testing.T) {
// source simulates an apiserver object endpoint.
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
// Make a controller that just panic if the AddFunc is called.
_, controller := NewInformer(
@@ -425,9 +431,10 @@ func TestPanicPropagated(t *testing.T) {
},
)
// Run the controller and run it until we close stop.
stop := make(chan struct{})
defer close(stop)
// Run the controller and run it until we cancel.
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancel(ctx)
defer cancel()
propagated := make(chan interface{})
go func() {
@@ -436,7 +443,7 @@ func TestPanicPropagated(t *testing.T) {
propagated <- r
}
}()
controller.Run(stop)
controller.RunWithContext(ctx)
}()
// Let's add a object to the source. It will trigger a panic.
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "test"}})
@@ -456,7 +463,7 @@ func TestPanicPropagated(t *testing.T) {
func TestTransformingInformer(t *testing.T) {
// source simulates an apiserver object endpoint.
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
makePod := func(name, generation string) *v1.Pod {
return &v1.Pod{
@@ -553,8 +560,10 @@ func TestTransformingInformer(t *testing.T) {
}
}
stopCh := make(chan struct{})
go controller.Run(stopCh)
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancel(ctx)
defer cancel()
go controller.RunWithContext(ctx)
verifyEvent(watch.Added, nil, expectedPod("pod1", "2"))
verifyStore([]interface{}{expectedPod("pod1", "2")})
@@ -572,13 +581,19 @@ func TestTransformingInformer(t *testing.T) {
source.Delete(makePod("pod1", "2"))
verifyEvent(watch.Deleted, expectedPod("pod1", "2"), nil)
verifyStore([]interface{}{expectedPod("pod2", "2"), expectedPod("pod3", "1")})
close(stopCh)
}
func TestTransformingInformerRace(t *testing.T) {
_, ctx := ktesting.NewTestContext(t)
// Canceled *only* when the test is done.
testCtx, cancel := context.WithCancel(ctx)
defer cancel()
// Canceled *also* during the test.
ctx, cancel = context.WithCancel(ctx)
defer cancel()
// source simulates an apiserver object endpoint.
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
label := "to-be-transformed"
makePod := func(name string) *v1.Pod {
@@ -616,7 +631,11 @@ func TestTransformingInformerRace(t *testing.T) {
type event struct{}
events := make(chan event, numObjs)
recordEvent := func(eventType watch.EventType, previous, current interface{}) {
events <- event{}
select {
case events <- event{}:
case <-testCtx.Done():
// Don't block forever in the write above when test is already done.
}
}
checkEvents := func(count int) {
for i := 0; i < count; i++ {
@@ -635,8 +654,7 @@ func TestTransformingInformerRace(t *testing.T) {
podTransformer,
)
stopCh := make(chan struct{})
go controller.Run(stopCh)
go controller.RunWithContext(ctx)
checkEvents(numObjs)
@@ -650,7 +668,7 @@ func TestTransformingInformerRace(t *testing.T) {
key := fmt.Sprintf("namespace/pod-%d", index)
for {
select {
case <-stopCh:
case <-ctx.Done():
return
default:
}
@@ -672,7 +690,7 @@ func TestTransformingInformerRace(t *testing.T) {
// Let resyncs to happen for some time.
time.Sleep(time.Second)
close(stopCh)
cancel()
wg.Wait()
close(errors)
for err := range errors {

25
staging/src/k8s.io/client-go/tools/cache/delta_fifo.go vendored
View File

@@ -55,6 +55,9 @@ type DeltaFIFOOptions struct {
// If set, will be called for objects before enqueueing them. Please
// see the comment on TransformFunc for details.
Transformer TransformFunc
// If set, log output will go to this logger instead of klog.Background().
Logger *klog.Logger
}
// DeltaFIFO is like FIFO, but differs in two ways. One is that the
@@ -136,6 +139,10 @@ type DeltaFIFO struct {
// Called with every object if non-nil.
transformer TransformFunc
// logger is a per-instance logger. This gets chosen when constructing
// the instance, with klog.Background() as default.
logger klog.Logger
}
// TransformFunc allows for transforming an object before it will be processed.
@@ -253,6 +260,10 @@ func NewDeltaFIFOWithOptions(opts DeltaFIFOOptions) *DeltaFIFO {
emitDeltaTypeReplaced: opts.EmitDeltaTypeReplaced,
transformer: opts.Transformer,
logger: klog.Background(),
}
if opts.Logger != nil {
f.logger = *opts.Logger
}
f.cond.L = &f.lock
return f
@@ -487,10 +498,10 @@ func (f *DeltaFIFO) queueActionInternalLocked(actionType, internalActionType Del
// when given a non-empty list (as it is here).
// If somehow it happens anyway, deal with it but complain.
if oldDeltas == nil {
klog.Errorf("Impossible dedupDeltas for id=%q: oldDeltas=%#+v, obj=%#+v; ignoring", id, oldDeltas, obj)
f.logger.Error(nil, "Impossible dedupDeltas, ignoring", "id", id, "oldDeltas", oldDeltas, "obj", obj)
return nil
}
klog.Errorf("Impossible dedupDeltas for id=%q: oldDeltas=%#+v, obj=%#+v; breaking invariant by storing empty Deltas", id, oldDeltas, obj)
f.logger.Error(nil, "Impossible dedupDeltas, breaking invariant by storing empty Deltas", "id", id, "oldDeltas", oldDeltas, "obj", obj)
f.items[id] = newDeltas
return fmt.Errorf("Impossible dedupDeltas for id=%q: oldDeltas=%#+v, obj=%#+v; broke DeltaFIFO invariant by storing empty Deltas", id, oldDeltas, obj)
}
@@ -597,7 +608,7 @@ func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
item, ok := f.items[id]
if !ok {
// This should never happen
klog.Errorf("Inconceivable! %q was in f.queue but not f.items; ignoring.", id)
f.logger.Error(nil, "Inconceivable! Item was in f.queue but not f.items; ignoring", "id", id)
continue
}
delete(f.items, id)
@@ -694,10 +705,10 @@ func (f *DeltaFIFO) Replace(list []interface{}, _ string) error {
deletedObj, exists, err := f.knownObjects.GetByKey(k)
if err != nil {
deletedObj = nil
klog.Errorf("Unexpected error %v during lookup of key %v, placing DeleteFinalStateUnknown marker without object", err, k)
f.logger.Error(err, "Unexpected error during lookup, placing DeleteFinalStateUnknown marker without object", "key", k)
} else if !exists {
deletedObj = nil
klog.Infof("Key %v does not exist in known objects store, placing DeleteFinalStateUnknown marker without object", k)
f.logger.Info("Key does not exist in known objects store, placing DeleteFinalStateUnknown marker without object", "key", k)
}
queuedDeletions++
if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil {
@@ -737,10 +748,10 @@ func (f *DeltaFIFO) Resync() error {
func (f *DeltaFIFO) syncKeyLocked(key string) error {
obj, exists, err := f.knownObjects.GetByKey(key)
if err != nil {
klog.Errorf("Unexpected error %v during lookup of key %v, unable to queue object for sync", err, key)
f.logger.Error(err, "Unexpected error during lookup, unable to queue object for sync", "key", key)
return nil
} else if !exists {
klog.Infof("Key %v does not exist in known objects store, unable to queue object for sync", key)
f.logger.Info("Key does not exist in known objects store, unable to queue object for sync", "key", key)
return nil
}

7
staging/src/k8s.io/client-go/tools/cache/listers.go vendored
View File

@@ -62,7 +62,12 @@ func ListAllByNamespace(indexer Indexer, namespace string, selector labels.Selec
items, err := indexer.Index(NamespaceIndex, &metav1.ObjectMeta{Namespace: namespace})
if err != nil {
// Ignore error; do slow search without index.
klog.Warningf("can not retrieve list of objects using index : %v", err)
//
// ListAllByNamespace is called by generated code
// (k8s.io/client-go/listers) and probably not worth converting
// to contextual logging, which would require changing all of
// those APIs.
klog.TODO().Info("Warning: can not retrieve list of objects using index", "err", err)
for _, m := range indexer.List() {
metadata, err := meta.Accessor(m)
if err != nil {

13
staging/src/k8s.io/client-go/tools/cache/main_test.go vendored
View File

@@ -17,10 +17,19 @@ limitations under the License.
package cache
import (
"os"
"testing"
"go.uber.org/goleak"
)
func TestMain(m *testing.M) {
os.Exit(m.Run())
options := []goleak.Option{
// These tests run goroutines which get stuck in Pop.
// This cannot be fixed without modifying the API.
goleak.IgnoreAnyFunction("k8s.io/client-go/tools/cache.TestFIFO_addReplace.func1"),
goleak.IgnoreAnyFunction("k8s.io/client-go/tools/cache.TestFIFO_addUpdate.func1"),
goleak.IgnoreAnyFunction("k8s.io/client-go/tools/cache.TestDeltaFIFO_addReplace.func1"),
goleak.IgnoreAnyFunction("k8s.io/client-go/tools/cache.TestDeltaFIFO_addUpdate.func1"),
}
goleak.VerifyTestMain(m, options...)
}

8
staging/src/k8s.io/client-go/tools/cache/mutation_cache.go vendored
View File

@@ -60,7 +60,7 @@ type ResourceVersionComparator interface {
// If includeAdds is true, objects in the mutation cache will be returned even if they don't exist
// in the underlying store. This is only safe if your use of the cache can handle mutation entries
// remaining in the cache for up to ttl when mutations and deletes occur very closely in time.
func NewIntegerResourceVersionMutationCache(backingCache Store, indexer Indexer, ttl time.Duration, includeAdds bool) MutationCache {
func NewIntegerResourceVersionMutationCache(logger klog.Logger, backingCache Store, indexer Indexer, ttl time.Duration, includeAdds bool) MutationCache {
return &mutationCache{
backingCache: backingCache,
indexer: indexer,
@@ -68,6 +68,7 @@ func NewIntegerResourceVersionMutationCache(backingCache Store, indexer Indexer,
comparator: etcdObjectVersioner{},
ttl: ttl,
includeAdds: includeAdds,
logger: logger,
}
}
@@ -75,6 +76,7 @@ func NewIntegerResourceVersionMutationCache(backingCache Store, indexer Indexer,
// since you can't distinguish between, "didn't observe create" and "was deleted after create",
// if the key is missing from the backing cache, we always return it as missing
type mutationCache struct {
logger klog.Logger
lock sync.Mutex
backingCache Store
indexer Indexer
@@ -157,7 +159,7 @@ func (c *mutationCache) ByIndex(name string, indexKey string) ([]interface{}, er
}
elements, err := fn(updated)
if err != nil {
klog.V(4).Infof("Unable to calculate an index entry for mutation cache entry %s: %v", key, err)
c.logger.V(4).Info("Unable to calculate an index entry for mutation cache entry", "key", key, "err", err)
continue
}
for _, inIndex := range elements {
@@ -204,7 +206,7 @@ func (c *mutationCache) Mutation(obj interface{}) {
key, err := DeletionHandlingMetaNamespaceKeyFunc(obj)
if err != nil {
// this is a "nice to have", so failures shouldn't do anything weird
utilruntime.HandleError(err)
utilruntime.HandleErrorWithLogger(c.logger, err, "DeletionHandlingMetaNamespaceKeyFunc")
return
}

1
staging/src/k8s.io/client-go/tools/cache/mutation_detector.go vendored
View File

@@ -50,6 +50,7 @@ func NewCacheMutationDetector(name string) MutationDetector {
if !mutationDetectionEnabled {
return dummyMutationDetector{}
}
//nolint:logcheck // This code shouldn't be used in production.
klog.Warningln("Mutation detector is enabled, this will result in memory leakage.")
return &defaultCacheMutationDetector{name: name, period: 1 * time.Second, retainDuration: 2 * time.Minute}
}

3
staging/src/k8s.io/client-go/tools/cache/processor_listener_test.go vendored
View File

@@ -22,6 +22,7 @@ import (
"time"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/klog/v2"
)
const (
@@ -35,7 +36,7 @@ func BenchmarkListener(b *testing.B) {
swg.Add(b.N)
b.SetParallelism(concurrencyLevel)
// Preallocate enough space so that benchmark does not run out of it
pl := newProcessListener(&ResourceEventHandlerFuncs{
pl := newProcessListener(klog.Background(), &ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
swg.Done()
},

183
staging/src/k8s.io/client-go/tools/cache/reflector.go vendored
View File

@@ -95,7 +95,7 @@ type Reflector struct {
// lastSyncResourceVersionMutex guards read/write access to lastSyncResourceVersion
lastSyncResourceVersionMutex sync.RWMutex
// Called whenever the ListAndWatch drops the connection with an error.
watchErrorHandler WatchErrorHandler
watchErrorHandler WatchErrorHandlerWithContext
// WatchListPageSize is the requested chunk size of initial and resync watch lists.
// If unset, for consistent reads (RV="") or reads that opt-into arbitrarily old data
// (RV="0") it will default to pager.PageSize, for the rest (RV != "" && RV != "0")
@@ -150,20 +150,32 @@ type ResourceVersionUpdater interface {
// should be offloaded.
type WatchErrorHandler func(r *Reflector, err error)
// DefaultWatchErrorHandler is the default implementation of WatchErrorHandler
func DefaultWatchErrorHandler(r *Reflector, err error) {
// The WatchErrorHandler is called whenever ListAndWatch drops the
// connection with an error. After calling this handler, the informer
// will backoff and retry.
//
// The default implementation looks at the error type and tries to log
// the error message at an appropriate level.
//
// Implementations of this handler may display the error message in other
// ways. Implementations should return quickly - any expensive processing
// should be offloaded.
type WatchErrorHandlerWithContext func(ctx context.Context, r *Reflector, err error)
// DefaultWatchErrorHandler is the default implementation of WatchErrorHandlerWithContext.
func DefaultWatchErrorHandler(ctx context.Context, r *Reflector, err error) {
switch {
case isExpiredError(err):
// Don't set LastSyncResourceVersionUnavailable - LIST call with ResourceVersion=RV already
// has a semantic that it returns data at least as fresh as provided RV.
// So first try to LIST with setting RV to resource version of last observed object.
klog.V(4).Infof("%s: watch of %v closed with: %v", r.name, r.typeDescription, err)
klog.FromContext(ctx).V(4).Info("Watch closed", "reflector", r.name, "type", r.typeDescription, "err", err)
case err == io.EOF:
// watch closed normally
case err == io.ErrUnexpectedEOF:
klog.V(1).Infof("%s: Watch for %v closed with unexpected EOF: %v", r.name, r.typeDescription, err)
klog.FromContext(ctx).V(1).Info("Watch closed with unexpected EOF", "reflector", r.name, "type", r.typeDescription, "err", err)
default:
utilruntime.HandleError(fmt.Errorf("%s: Failed to watch %v: %v", r.name, r.typeDescription, err))
utilruntime.HandleErrorWithContext(ctx, err, "Failed to watch", "reflector", r.name, "type", r.typeDescription)
}
}
@@ -243,7 +255,7 @@ func NewReflectorWithOptions(lw ListerWatcher, expectedType interface{}, store S
// 0.22 QPS. If we don't backoff for 2min, assume API server is healthy and we reset the backoff.
backoffManager: wait.NewExponentialBackoffManager(800*time.Millisecond, 30*time.Second, 2*time.Minute, 2.0, 1.0, reflectorClock),
clock: reflectorClock,
watchErrorHandler: WatchErrorHandler(DefaultWatchErrorHandler),
watchErrorHandler: WatchErrorHandlerWithContext(DefaultWatchErrorHandler),
expectedType: reflect.TypeOf(expectedType),
}
@@ -309,14 +321,24 @@ var internalPackages = []string{"client-go/tools/cache/"}
// Run repeatedly uses the reflector's ListAndWatch to fetch all the
// objects and subsequent deltas.
// Run will exit when stopCh is closed.
//
// Contextual logging: RunWithContext should be used instead of Run in code which supports contextual logging.
func (r *Reflector) Run(stopCh <-chan struct{}) {
klog.V(3).Infof("Starting reflector %s (%s) from %s", r.typeDescription, r.resyncPeriod, r.name)
r.RunWithContext(wait.ContextForChannel(stopCh))
}
// RunWithContext repeatedly uses the reflector's ListAndWatch to fetch all the
// objects and subsequent deltas.
// Run will exit when the context is canceled.
func (r *Reflector) RunWithContext(ctx context.Context) {
logger := klog.FromContext(ctx)
logger.V(3).Info("Starting reflector", "type", r.typeDescription, "resyncPeriod", r.resyncPeriod, "reflector", r.name)
wait.BackoffUntil(func() {
if err := r.ListAndWatch(stopCh); err != nil {
r.watchErrorHandler(r, err)
if err := r.ListAndWatchWithContext(ctx); err != nil {
r.watchErrorHandler(ctx, r, err)
}
}, r.backoffManager, true, stopCh)
klog.V(3).Infof("Stopping reflector %s (%s) from %s", r.typeDescription, r.resyncPeriod, r.name)
}, r.backoffManager, true, ctx.Done())
logger.V(3).Info("Stopping reflector", "type", r.typeDescription, "resyncPeriod", r.resyncPeriod, "reflector", r.name)
}
var (
@@ -345,21 +367,31 @@ func (r *Reflector) resyncChan() (<-chan time.Time, func() bool) {
// ListAndWatch first lists all items and get the resource version at the moment of call,
// and then use the resource version to watch.
// It returns error if ListAndWatch didn't even try to initialize watch.
//
// Contextual logging: ListAndWatchWithContext should be used instead of ListAndWatch in code which supports contextual logging.
func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
klog.V(3).Infof("Listing and watching %v from %s", r.typeDescription, r.name)
return r.ListAndWatchWithContext(wait.ContextForChannel(stopCh))
}
// ListAndWatchWithContext first lists all items and get the resource version at the moment of call,
// and then use the resource version to watch.
// It returns error if ListAndWatchWithContext didn't even try to initialize watch.
func (r *Reflector) ListAndWatchWithContext(ctx context.Context) error {
logger := klog.FromContext(ctx)
logger.V(3).Info("Listing and watching", "type", r.typeDescription, "reflector", r.name)
var err error
var w watch.Interface
useWatchList := ptr.Deref(r.UseWatchList, false)
fallbackToList := !useWatchList
if useWatchList {
w, err = r.watchList(stopCh)
w, err = r.watchList(ctx)
if w == nil && err == nil {
// stopCh was closed
return nil
}
if err != nil {
klog.Warningf("The watchlist request ended with an error, falling back to the standard LIST/WATCH semantics because making progress is better than deadlocking, err = %v", err)
logger.Error(err, "The watchlist request ended with an error, falling back to the standard LIST/WATCH semantics because making progress is better than deadlocking")
fallbackToList = true
// ensure that we won't accidentally pass some garbage down the watch.
w = nil
@@ -367,20 +399,21 @@ func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
}
if fallbackToList {
err = r.list(stopCh)
err = r.list(ctx)
if err != nil {
return err
}
}
klog.V(2).Infof("Caches populated for %v from %s", r.typeDescription, r.name)
return r.watchWithResync(w, stopCh)
logger.V(2).Info("Caches populated", "type", r.typeDescription, "reflector", r.name)
return r.watchWithResync(ctx, w)
}
// startResync periodically calls r.store.Resync() method.
// Note that this method is blocking and should be
// called in a separate goroutine.
func (r *Reflector) startResync(stopCh <-chan struct{}, cancelCh <-chan struct{}, resyncerrc chan error) {
func (r *Reflector) startResync(ctx context.Context, resyncerrc chan error) {
logger := klog.FromContext(ctx)
resyncCh, cleanup := r.resyncChan()
defer func() {
cleanup() // Call the last one written into cleanup
@@ -388,13 +421,11 @@ func (r *Reflector) startResync(stopCh <-chan struct{}, cancelCh <-chan struct{}
for {
select {
case <-resyncCh:
case <-stopCh:
return
case <-cancelCh:
case <-ctx.Done():
return
}
if r.ShouldResync == nil || r.ShouldResync() {
klog.V(4).Infof("%s: forcing resync", r.name)
logger.V(4).Info("Forcing resync", "reflector", r.name)
if err := r.store.Resync(); err != nil {
resyncerrc <- err
return
@@ -406,16 +437,27 @@ func (r *Reflector) startResync(stopCh <-chan struct{}, cancelCh <-chan struct{}
}
// watchWithResync runs watch with startResync in the background.
func (r *Reflector) watchWithResync(w watch.Interface, stopCh <-chan struct{}) error {
func (r *Reflector) watchWithResync(ctx context.Context, w watch.Interface) error {
resyncerrc := make(chan error, 1)
cancelCh := make(chan struct{})
defer close(cancelCh)
go r.startResync(stopCh, cancelCh, resyncerrc)
return r.watch(w, stopCh, resyncerrc)
cancelCtx, cancel := context.WithCancel(ctx)
// Waiting for completion of the goroutine is relevant for race detector.
// Without this, there is a race between "this function returns + code
// waiting for it" and "goroutine does something".
var wg wait.Group
defer func() {
cancel()
wg.Wait()
}()
wg.Start(func() {
r.startResync(cancelCtx, resyncerrc)
})
return r.watch(ctx, w, resyncerrc)
}
// watch simply starts a watch request with the server.
func (r *Reflector) watch(w watch.Interface, stopCh <-chan struct{}, resyncerrc chan error) error {
func (r *Reflector) watch(ctx context.Context, w watch.Interface, resyncerrc chan error) error {
stopCh := ctx.Done()
logger := klog.FromContext(ctx)
var err error
retry := NewRetryWithDeadline(r.MaxInternalErrorRetryDuration, time.Minute, apierrors.IsInternalError, r.clock)
@@ -451,7 +493,7 @@ func (r *Reflector) watch(w watch.Interface, stopCh <-chan struct{}, resyncerrc
w, err = r.listerWatcher.Watch(options)
if err != nil {
if canRetry := isWatchErrorRetriable(err); canRetry {
klog.V(4).Infof("%s: watch of %v returned %v - backing off", r.name, r.typeDescription, err)
logger.V(4).Info("Watch failed - backing off", "reflector", r.name, "type", r.typeDescription, "err", err)
select {
case <-stopCh:
return nil
@@ -463,8 +505,8 @@ func (r *Reflector) watch(w watch.Interface, stopCh <-chan struct{}, resyncerrc
}
}
err = handleWatch(start, w, r.store, r.expectedType, r.expectedGVK, r.name, r.typeDescription, r.setLastSyncResourceVersion,
r.clock, resyncerrc, stopCh)
err = handleWatch(ctx, start, w, r.store, r.expectedType, r.expectedGVK, r.name, r.typeDescription, r.setLastSyncResourceVersion,
r.clock, resyncerrc)
// Ensure that watch will not be reused across iterations.
w.Stop()
w = nil
@@ -476,9 +518,9 @@ func (r *Reflector) watch(w watch.Interface, stopCh <-chan struct{}, resyncerrc
// Don't set LastSyncResourceVersionUnavailable - LIST call with ResourceVersion=RV already
// has a semantic that it returns data at least as fresh as provided RV.
// So first try to LIST with setting RV to resource version of last observed object.
klog.V(4).Infof("%s: watch of %v closed with: %v", r.name, r.typeDescription, err)
logger.V(4).Info("Watch closed", "reflector", r.name, "type", r.typeDescription, "err", err)
case apierrors.IsTooManyRequests(err):
klog.V(2).Infof("%s: watch of %v returned 429 - backing off", r.name, r.typeDescription)
logger.V(2).Info("Watch returned 429 - backing off", "reflector", r.name, "type", r.typeDescription)
select {
case <-stopCh:
return nil
@@ -486,10 +528,10 @@ func (r *Reflector) watch(w watch.Interface, stopCh <-chan struct{}, resyncerrc
continue
}
case apierrors.IsInternalError(err) && retry.ShouldRetry():
klog.V(2).Infof("%s: retrying watch of %v internal error: %v", r.name, r.typeDescription, err)
logger.V(2).Info("Retrying watch after internal error", "reflector", r.name, "type", r.typeDescription, "err", err)
continue
default:
klog.Warningf("%s: watch of %v ended with: %v", r.name, r.typeDescription, err)
logger.Info("Warning: watch ended with error", "reflector", r.name, "type", r.typeDescription, "err", err)
}
}
return nil
@@ -499,7 +541,7 @@ func (r *Reflector) watch(w watch.Interface, stopCh <-chan struct{}, resyncerrc
// list simply lists all items and records a resource version obtained from the server at the moment of the call.
// the resource version can be used for further progress notification (aka. watch).
func (r *Reflector) list(stopCh <-chan struct{}) error {
func (r *Reflector) list(ctx context.Context) error {
var resourceVersion string
options := metav1.ListOptions{ResourceVersion: r.relistResourceVersion()}
@@ -558,7 +600,7 @@ func (r *Reflector) list(stopCh <-chan struct{}) error {
close(listCh)
}()
select {
case <-stopCh:
case <-ctx.Done():
return nil
case r := <-panicCh:
panic(r)
@@ -566,7 +608,6 @@ func (r *Reflector) list(stopCh <-chan struct{}) error {
}
initTrace.Step("Objects listed", trace.Field{Key: "error", Value: err})
if err != nil {
klog.Warningf("%s: failed to list %v: %v", r.name, r.typeDescription, err)
return fmt.Errorf("failed to list %v: %w", r.typeDescription, err)
}
@@ -624,7 +665,9 @@ func (r *Reflector) list(stopCh <-chan struct{}) error {
// After receiving a "Bookmark" event the reflector is considered to be synchronized.
// It replaces its internal store with the collected items and
// reuses the current watch requests for getting further events.
func (r *Reflector) watchList(stopCh <-chan struct{}) (watch.Interface, error) {
func (r *Reflector) watchList(ctx context.Context) (watch.Interface, error) {
stopCh := ctx.Done()
logger := klog.FromContext(ctx)
var w watch.Interface
var err error
var temporaryStore Store
@@ -634,7 +677,7 @@ func (r *Reflector) watchList(stopCh <-chan struct{}) (watch.Interface, error) {
// could be unified with the r.watch method
isErrorRetriableWithSideEffectsFn := func(err error) bool {
if canRetry := isWatchErrorRetriable(err); canRetry {
klog.V(2).Infof("%s: watch-list of %v returned %v - backing off", r.name, r.typeDescription, err)
logger.V(2).Info("watch-list failed - backing off", "reflector", r.name, "type", r.typeDescription, "err", err)
<-r.backoffManager.Backoff().C()
return true
}
@@ -681,9 +724,9 @@ func (r *Reflector) watchList(stopCh <-chan struct{}) (watch.Interface, error) {
}
return nil, err
}
watchListBookmarkReceived, err := handleListWatch(start, w, temporaryStore, r.expectedType, r.expectedGVK, r.name, r.typeDescription,
watchListBookmarkReceived, err := handleListWatch(ctx, start, w, temporaryStore, r.expectedType, r.expectedGVK, r.name, r.typeDescription,
func(rv string) { resourceVersion = rv },
r.clock, make(chan error), stopCh)
r.clock, make(chan error))
if err != nil {
w.Stop() // stop and retry with clean state
if errors.Is(err, errorStopRequested) {
@@ -706,7 +749,7 @@ func (r *Reflector) watchList(stopCh <-chan struct{}) (watch.Interface, error) {
// we utilize the temporaryStore to ensure independence from the current store implementation.
// as of today, the store is implemented as a queue and will be drained by the higher-level
// component as soon as it finishes replacing the content.
checkWatchListDataConsistencyIfRequested(wait.ContextForChannel(stopCh), r.name, resourceVersion, wrapListFuncWithContext(r.listerWatcher.List), temporaryStore.List)
checkWatchListDataConsistencyIfRequested(ctx, r.name, resourceVersion, wrapListFuncWithContext(r.listerWatcher.List), temporaryStore.List)
if err := r.store.Replace(temporaryStore.List(), resourceVersion); err != nil {
return nil, fmt.Errorf("unable to sync watch-list result: %w", err)
@@ -731,6 +774,7 @@ func (r *Reflector) syncWith(items []runtime.Object, resourceVersion string) err
// retry. If successful, the watcher will be left open after receiving the
// initial set of objects, to allow watching for future events.
func handleListWatch(
ctx context.Context,
start time.Time,
w watch.Interface,
store Store,
@@ -741,17 +785,17 @@ func handleListWatch(
setLastSyncResourceVersion func(string),
clock clock.Clock,
errCh chan error,
stopCh <-chan struct{},
) (bool, error) {
exitOnWatchListBookmarkReceived := true
return handleAnyWatch(start, w, store, expectedType, expectedGVK, name, expectedTypeName,
setLastSyncResourceVersion, exitOnWatchListBookmarkReceived, clock, errCh, stopCh)
return handleAnyWatch(ctx, start, w, store, expectedType, expectedGVK, name, expectedTypeName,
setLastSyncResourceVersion, exitOnWatchListBookmarkReceived, clock, errCh)
}
// handleListWatch consumes events from w, updates the Store, and records the
// last seen ResourceVersion, to allow continuing from that ResourceVersion on
// retry. The watcher will always be stopped on exit.
func handleWatch(
ctx context.Context,
start time.Time,
w watch.Interface,
store Store,
@@ -762,11 +806,10 @@ func handleWatch(
setLastSyncResourceVersion func(string),
clock clock.Clock,
errCh chan error,
stopCh <-chan struct{},
) error {
exitOnWatchListBookmarkReceived := false
_, err := handleAnyWatch(start, w, store, expectedType, expectedGVK, name, expectedTypeName,
setLastSyncResourceVersion, exitOnWatchListBookmarkReceived, clock, errCh, stopCh)
_, err := handleAnyWatch(ctx, start, w, store, expectedType, expectedGVK, name, expectedTypeName,
setLastSyncResourceVersion, exitOnWatchListBookmarkReceived, clock, errCh)
return err
}
@@ -779,7 +822,9 @@ func handleWatch(
// The watcher will always be stopped, unless exitOnWatchListBookmarkReceived is
// true and watchListBookmarkReceived is true. This allows the same watch stream
// to be re-used by the caller to continue watching for new events.
func handleAnyWatch(start time.Time,
func handleAnyWatch(
ctx context.Context,
start time.Time,
w watch.Interface,
store Store,
expectedType reflect.Type,
@@ -790,17 +835,17 @@ func handleAnyWatch(start time.Time,
exitOnWatchListBookmarkReceived bool,
clock clock.Clock,
errCh chan error,
stopCh <-chan struct{},
) (bool, error) {
watchListBookmarkReceived := false
eventCount := 0
initialEventsEndBookmarkWarningTicker := newInitialEventsEndBookmarkTicker(name, clock, start, exitOnWatchListBookmarkReceived)
logger := klog.FromContext(ctx)
initialEventsEndBookmarkWarningTicker := newInitialEventsEndBookmarkTicker(logger, name, clock, start, exitOnWatchListBookmarkReceived)
defer initialEventsEndBookmarkWarningTicker.Stop()
loop:
for {
select {
case <-stopCh:
case <-ctx.Done():
return watchListBookmarkReceived, errorStopRequested
case err := <-errCh:
return watchListBookmarkReceived, err
@@ -813,19 +858,19 @@ loop:
}
if expectedType != nil {
if e, a := expectedType, reflect.TypeOf(event.Object); e != a {
utilruntime.HandleError(fmt.Errorf("%s: expected type %v, but watch event object had type %v", name, e, a))
utilruntime.HandleErrorWithContext(ctx, nil, "Unexpected watch event object type", "reflector", name, "expectedType", e, "actualType", a)
continue
}
}
if expectedGVK != nil {
if e, a := *expectedGVK, event.Object.GetObjectKind().GroupVersionKind(); e != a {
utilruntime.HandleError(fmt.Errorf("%s: expected gvk %v, but watch event object had gvk %v", name, e, a))
utilruntime.HandleErrorWithContext(ctx, nil, "Unexpected watch event object gvk", "reflector", name, "expectedGVK", e, "actualGVK", a)
continue
}
}
meta, err := meta.Accessor(event.Object)
if err != nil {
utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", name, event))
utilruntime.HandleErrorWithContext(ctx, err, "Unable to understand watch event", "reflector", name, "event", event)
continue
}
resourceVersion := meta.GetResourceVersion()
@@ -833,12 +878,12 @@ loop:
case watch.Added:
err := store.Add(event.Object)
if err != nil {
utilruntime.HandleError(fmt.Errorf("%s: unable to add watch event object (%#v) to store: %v", name, event.Object, err))
utilruntime.HandleErrorWithContext(ctx, err, "Unable to add watch event object to store", "reflector", name, "object", event.Object)
}
case watch.Modified:
err := store.Update(event.Object)
if err != nil {
utilruntime.HandleError(fmt.Errorf("%s: unable to update watch event object (%#v) to store: %v", name, event.Object, err))
utilruntime.HandleErrorWithContext(ctx, err, "Unable to update watch event object to store", "reflector", name, "object", event.Object)
}
case watch.Deleted:
// TODO: Will any consumers need access to the "last known
@@ -846,7 +891,7 @@ loop:
// to change this.
err := store.Delete(event.Object)
if err != nil {
utilruntime.HandleError(fmt.Errorf("%s: unable to delete watch event object (%#v) from store: %v", name, event.Object, err))
utilruntime.HandleErrorWithContext(ctx, err, "Unable to delete watch event object from store", "reflector", name, "object", event.Object)
}
case watch.Bookmark:
// A `Bookmark` means watch has synced here, just update the resourceVersion
@@ -854,7 +899,7 @@ loop:
watchListBookmarkReceived = true
}
default:
utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", name, event))
utilruntime.HandleErrorWithContext(ctx, err, "Unknown watch event", "reflector", name, "event", event)
}
setLastSyncResourceVersion(resourceVersion)
if rvu, ok := store.(ResourceVersionUpdater); ok {
@@ -863,7 +908,7 @@ loop:
eventCount++
if exitOnWatchListBookmarkReceived && watchListBookmarkReceived {
watchDuration := clock.Since(start)
klog.V(4).Infof("exiting %v Watch because received the bookmark that marks the end of initial events stream, total %v items received in %v", name, eventCount, watchDuration)
klog.FromContext(ctx).V(4).Info("Exiting watch because received the bookmark that marks the end of initial events stream", "reflector", name, "totalItems", eventCount, "duration", watchDuration)
return watchListBookmarkReceived, nil
}
initialEventsEndBookmarkWarningTicker.observeLastEventTimeStamp(clock.Now())
@@ -876,7 +921,7 @@ loop:
if watchDuration < 1*time.Second && eventCount == 0 {
return watchListBookmarkReceived, fmt.Errorf("very short watch: %s: Unexpected watch close - watch lasted less than a second and no items received", name)
}
klog.V(4).Infof("%s: Watch close - %v total %v items received", name, expectedTypeName, eventCount)
klog.FromContext(ctx).V(4).Info("Watch close", "reflector", name, "type", expectedTypeName, "totalItems", eventCount)
return watchListBookmarkReceived, nil
}
@@ -1006,6 +1051,7 @@ type initialEventsEndBookmarkTicker struct {
clock.Ticker
clock clock.Clock
name string
logger klog.Logger
watchStart time.Time
tickInterval time.Duration
@@ -1019,15 +1065,15 @@ type initialEventsEndBookmarkTicker struct {
// Note that the caller controls whether to call t.C() and t.Stop().
//
// In practice, the reflector exits the watchHandler as soon as the bookmark event is received and calls the t.C() method.
func newInitialEventsEndBookmarkTicker(name string, c clock.Clock, watchStart time.Time, exitOnWatchListBookmarkReceived bool) *initialEventsEndBookmarkTicker {
return newInitialEventsEndBookmarkTickerInternal(name, c, watchStart, 10*time.Second, exitOnWatchListBookmarkReceived)
func newInitialEventsEndBookmarkTicker(logger klog.Logger, name string, c clock.Clock, watchStart time.Time, exitOnWatchListBookmarkReceived bool) *initialEventsEndBookmarkTicker {
return newInitialEventsEndBookmarkTickerInternal(logger, name, c, watchStart, 10*time.Second, exitOnWatchListBookmarkReceived)
}
func newInitialEventsEndBookmarkTickerInternal(name string, c clock.Clock, watchStart time.Time, tickInterval time.Duration, exitOnWatchListBookmarkReceived bool) *initialEventsEndBookmarkTicker {
func newInitialEventsEndBookmarkTickerInternal(logger klog.Logger, name string, c clock.Clock, watchStart time.Time, tickInterval time.Duration, exitOnWatchListBookmarkReceived bool) *initialEventsEndBookmarkTicker {
clockWithTicker, ok := c.(clock.WithTicker)
if !ok || !exitOnWatchListBookmarkReceived {
if exitOnWatchListBookmarkReceived {
klog.Warningf("clock does not support WithTicker interface but exitOnInitialEventsEndBookmark was requested")
logger.Info("Warning: clock does not support WithTicker interface but exitOnInitialEventsEndBookmark was requested")
}
return &initialEventsEndBookmarkTicker{
Ticker: &noopTicker{},
@@ -1038,6 +1084,7 @@ func newInitialEventsEndBookmarkTickerInternal(name string, c clock.Clock, watch
Ticker: clockWithTicker.NewTicker(tickInterval),
clock: c,
name: name,
logger: logger,
watchStart: watchStart,
tickInterval: tickInterval,
}
@@ -1049,7 +1096,7 @@ func (t *initialEventsEndBookmarkTicker) observeLastEventTimeStamp(lastEventObse
func (t *initialEventsEndBookmarkTicker) warnIfExpired() {
if err := t.produceWarningIfExpired(); err != nil {
klog.Warning(err)
t.logger.Info("Warning: event bookmark expired", "err", err)
}
}

207
staging/src/k8s.io/client-go/tools/cache/reflector_test.go vendored
View File

@@ -45,6 +45,7 @@ import (
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/klog/v2/ktesting"
"k8s.io/utils/clock"
testingclock "k8s.io/utils/clock/testing"
)
@@ -64,6 +65,7 @@ func (t *testLW) Watch(options metav1.ListOptions) (watch.Interface, error) {
}
func TestCloseWatchChannelOnError(t *testing.T) {
_, ctx := ktesting.NewTestContext(t)
r := NewReflector(&testLW{}, &v1.Pod{}, NewStore(MetaNamespaceKeyFunc), 0)
pod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "bar"}}
fw := watch.NewFake()
@@ -75,7 +77,7 @@ func TestCloseWatchChannelOnError(t *testing.T) {
return &v1.PodList{ListMeta: metav1.ListMeta{ResourceVersion: "1"}}, nil
},
}
go r.ListAndWatch(wait.NeverStop)
go func() { assert.NoError(t, r.ListAndWatchWithContext(ctx)) }()
fw.Error(pod)
select {
case _, ok := <-fw.ResultChan():
@@ -89,7 +91,8 @@ func TestCloseWatchChannelOnError(t *testing.T) {
}
func TestRunUntil(t *testing.T) {
stopCh := make(chan struct{})
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancelCause(ctx)
store := NewStore(MetaNamespaceKeyFunc)
r := NewReflector(&testLW{}, &v1.Pod{}, store, 0)
fw := watch.NewFake()
@@ -104,13 +107,13 @@ func TestRunUntil(t *testing.T) {
doneCh := make(chan struct{})
go func() {
defer close(doneCh)
r.Run(stopCh)
r.RunWithContext(ctx)
}()
// Synchronously add a dummy pod into the watch channel so we
// know the RunUntil go routine is in the watch handler.
fw.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "bar"}})
close(stopCh)
cancel(errors.New("done"))
resultCh := fw.ResultChan()
for {
select {
@@ -149,8 +152,9 @@ func TestReflectorResyncChan(t *testing.T) {
// TestReflectorWatchStoppedBefore ensures that neither List nor Watch are
// called if the stop channel is closed before Reflector.watch is called.
func TestReflectorWatchStoppedBefore(t *testing.T) {
stopCh := make(chan struct{})
close(stopCh)
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancelCause(ctx)
cancel(errors.New("don't run"))
lw := &ListWatch{
ListFunc: func(_ metav1.ListOptions) (runtime.Object, error) {
@@ -165,14 +169,15 @@ func TestReflectorWatchStoppedBefore(t *testing.T) {
}
target := NewReflector(lw, &v1.Pod{}, nil, 0)
err := target.watch(nil, stopCh, nil)
err := target.watch(ctx, nil, nil)
require.NoError(t, err)
}
// TestReflectorWatchStoppedAfter ensures that neither the watcher is stopped if
// the stop channel is closed after Reflector.watch has started watching.
func TestReflectorWatchStoppedAfter(t *testing.T) {
stopCh := make(chan struct{})
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancelCause(ctx)
var watchers []*watch.FakeWatcher
@@ -185,7 +190,7 @@ func TestReflectorWatchStoppedAfter(t *testing.T) {
// Simulate the stop channel being closed after watching has started
go func() {
time.Sleep(10 * time.Millisecond)
close(stopCh)
cancel(errors.New("10ms timeout reached"))
}()
// Use a fake watcher that never sends events
w := watch.NewFake()
@@ -195,7 +200,7 @@ func TestReflectorWatchStoppedAfter(t *testing.T) {
}
target := NewReflector(lw, &v1.Pod{}, nil, 0)
err := target.watch(nil, stopCh, nil)
err := target.watch(ctx, nil, nil)
require.NoError(t, err)
require.Len(t, watchers, 1)
require.True(t, watchers[0].IsStopped())
@@ -219,9 +224,10 @@ func BenchmarkReflectorResyncChanMany(b *testing.B) {
func TestReflectorHandleWatchStoppedBefore(t *testing.T) {
s := NewStore(MetaNamespaceKeyFunc)
g := NewReflector(&testLW{}, &v1.Pod{}, s, 0)
stopCh := make(chan struct{})
// Simulate the watch channel being closed before the watchHandler is called
close(stopCh)
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancelCause(ctx)
// Simulate the context being canceled before the watchHandler is called
cancel(errors.New("don't run"))
var calls []string
resultCh := make(chan watch.Event)
fw := watch.MockWatcher{
@@ -234,7 +240,7 @@ func TestReflectorHandleWatchStoppedBefore(t *testing.T) {
return resultCh
},
}
err := handleWatch(time.Now(), fw, s, g.expectedType, g.expectedGVK, g.name, g.typeDescription, g.setLastSyncResourceVersion, g.clock, nevererrc, stopCh)
err := handleWatch(ctx, time.Now(), fw, s, g.expectedType, g.expectedGVK, g.name, g.typeDescription, g.setLastSyncResourceVersion, g.clock, nevererrc)
if err == nil {
t.Errorf("unexpected non-error")
}
@@ -251,7 +257,8 @@ func TestReflectorHandleWatchStoppedAfter(t *testing.T) {
s := NewStore(MetaNamespaceKeyFunc)
g := NewReflector(&testLW{}, &v1.Pod{}, s, 0)
var calls []string
stopCh := make(chan struct{})
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancelCause(ctx)
resultCh := make(chan watch.Event)
fw := watch.MockWatcher{
StopFunc: func() {
@@ -265,12 +272,12 @@ func TestReflectorHandleWatchStoppedAfter(t *testing.T) {
// caller, after watching has started.
go func() {
time.Sleep(10 * time.Millisecond)
close(stopCh)
cancel(errors.New("10ms timeout reached"))
}()
return resultCh
},
}
err := handleWatch(time.Now(), fw, s, g.expectedType, g.expectedGVK, g.name, g.typeDescription, g.setLastSyncResourceVersion, g.clock, nevererrc, stopCh)
err := handleWatch(ctx, time.Now(), fw, s, g.expectedType, g.expectedGVK, g.name, g.typeDescription, g.setLastSyncResourceVersion, g.clock, nevererrc)
if err == nil {
t.Errorf("unexpected non-error")
}
@@ -285,6 +292,7 @@ func TestReflectorHandleWatchStoppedAfter(t *testing.T) {
func TestReflectorHandleWatchResultChanClosedBefore(t *testing.T) {
s := NewStore(MetaNamespaceKeyFunc)
g := NewReflector(&testLW{}, &v1.Pod{}, s, 0)
_, ctx := ktesting.NewTestContext(t)
var calls []string
resultCh := make(chan watch.Event)
fw := watch.MockWatcher{
@@ -298,7 +306,7 @@ func TestReflectorHandleWatchResultChanClosedBefore(t *testing.T) {
}
// Simulate the result channel being closed by the producer before handleWatch is called.
close(resultCh)
err := handleWatch(time.Now(), fw, s, g.expectedType, g.expectedGVK, g.name, g.typeDescription, g.setLastSyncResourceVersion, g.clock, nevererrc, wait.NeverStop)
err := handleWatch(ctx, time.Now(), fw, s, g.expectedType, g.expectedGVK, g.name, g.typeDescription, g.setLastSyncResourceVersion, g.clock, nevererrc)
if err == nil {
t.Errorf("unexpected non-error")
}
@@ -313,6 +321,7 @@ func TestReflectorHandleWatchResultChanClosedBefore(t *testing.T) {
func TestReflectorHandleWatchResultChanClosedAfter(t *testing.T) {
s := NewStore(MetaNamespaceKeyFunc)
g := NewReflector(&testLW{}, &v1.Pod{}, s, 0)
_, ctx := ktesting.NewTestContext(t)
var calls []string
resultCh := make(chan watch.Event)
fw := watch.MockWatcher{
@@ -331,7 +340,7 @@ func TestReflectorHandleWatchResultChanClosedAfter(t *testing.T) {
return resultCh
},
}
err := handleWatch(time.Now(), fw, s, g.expectedType, g.expectedGVK, g.name, g.typeDescription, g.setLastSyncResourceVersion, g.clock, nevererrc, wait.NeverStop)
err := handleWatch(ctx, time.Now(), fw, s, g.expectedType, g.expectedGVK, g.name, g.typeDescription, g.setLastSyncResourceVersion, g.clock, nevererrc)
if err == nil {
t.Errorf("unexpected non-error")
}
@@ -348,11 +357,12 @@ func TestReflectorWatchHandler(t *testing.T) {
// watching after all the events have been consumed. This avoids race
// conditions which can happen if the producer calls Stop(), instead of the
// consumer.
stopCh := make(chan struct{})
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancelCause(ctx)
setLastSyncResourceVersion := func(rv string) {
g.setLastSyncResourceVersion(rv)
if rv == "32" {
close(stopCh)
cancel(errors.New("LastSyncResourceVersion is 32"))
}
}
fw := watch.NewFake()
@@ -365,7 +375,7 @@ func TestReflectorWatchHandler(t *testing.T) {
fw.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "baz", ResourceVersion: "32"}})
fw.Stop()
}()
err := handleWatch(time.Now(), fw, s, g.expectedType, g.expectedGVK, g.name, g.typeDescription, setLastSyncResourceVersion, g.clock, nevererrc, stopCh)
err := handleWatch(ctx, time.Now(), fw, s, g.expectedType, g.expectedGVK, g.name, g.typeDescription, setLastSyncResourceVersion, g.clock, nevererrc)
// TODO(karlkfi): Fix FakeWatcher to avoid race condition between watcher.Stop() & close(stopCh)
if err != nil && !errors.Is(err, errorStopRequested) {
t.Errorf("unexpected error %v", err)
@@ -408,15 +418,19 @@ func TestReflectorStopWatch(t *testing.T) {
s := NewStore(MetaNamespaceKeyFunc)
g := NewReflector(&testLW{}, &v1.Pod{}, s, 0)
fw := watch.NewFake()
stopWatch := make(chan struct{})
close(stopWatch)
err := handleWatch(time.Now(), fw, s, g.expectedType, g.expectedGVK, g.name, g.typeDescription, g.setLastSyncResourceVersion, g.clock, nevererrc, stopWatch)
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancelCause(ctx)
cancel(errors.New("don't run"))
err := handleWatch(ctx, time.Now(), fw, s, g.expectedType, g.expectedGVK, g.name, g.typeDescription, g.setLastSyncResourceVersion, g.clock, nevererrc)
if err != errorStopRequested {
t.Errorf("expected stop error, got %q", err)
}
}
func TestReflectorListAndWatch(t *testing.T) {
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancel(ctx)
defer cancel()
createdFakes := make(chan *watch.FakeWatcher)
// The ListFunc says that it's at revision 1. Therefore, we expect our WatchFunc
@@ -442,7 +456,7 @@ func TestReflectorListAndWatch(t *testing.T) {
}
s := NewFIFO(MetaNamespaceKeyFunc)
r := NewReflector(lw, &v1.Pod{}, s, 0)
go r.ListAndWatch(wait.NeverStop)
go func() { assert.NoError(t, r.ListAndWatchWithContext(ctx)) }()
ids := []string{"foo", "bar", "baz", "qux", "zoo"}
var fw *watch.FakeWatcher
@@ -539,7 +553,8 @@ func TestReflectorListAndWatchWithErrors(t *testing.T) {
}
}
watchRet, watchErr := item.events, item.watchErr
stopCh := make(chan struct{})
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancelCause(ctx)
lw := &testLW{
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
if watchErr != nil {
@@ -557,7 +572,7 @@ func TestReflectorListAndWatchWithErrors(t *testing.T) {
// result channel, and wait for the consumer to stop the
// watcher, to avoid race conditions.
// TODO: Fix the FakeWatcher to separate watcher.Stop from close(resultCh)
close(stopCh)
cancel(errors.New("done"))
}()
return fw, nil
},
@@ -566,7 +581,7 @@ func TestReflectorListAndWatchWithErrors(t *testing.T) {
},
}
r := NewReflector(lw, &v1.Pod{}, s, 0)
err := r.ListAndWatch(stopCh)
err := r.ListAndWatchWithContext(ctx)
if item.listErr != nil && !errors.Is(err, item.listErr) {
t.Errorf("unexpected ListAndWatch error: %v", err)
}
@@ -593,7 +608,8 @@ func TestReflectorListAndWatchInitConnBackoff(t *testing.T) {
for _, test := range table {
t.Run(fmt.Sprintf("%d connection failures takes at least %d ms", test.numConnFails, 1<<test.numConnFails),
func(t *testing.T) {
stopCh := make(chan struct{})
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancelCause(ctx)
connFails := test.numConnFails
fakeClock := testingclock.NewFakeClock(time.Unix(0, 0))
bm := wait.NewExponentialBackoffManager(time.Millisecond, maxBackoff, 100*time.Millisecond, 2.0, 1.0, fakeClock)
@@ -624,7 +640,7 @@ func TestReflectorListAndWatchInitConnBackoff(t *testing.T) {
connFails--
return nil, syscall.ECONNREFUSED
}
close(stopCh)
cancel(errors.New("done"))
return watch.NewFake(), nil
},
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
@@ -637,10 +653,10 @@ func TestReflectorListAndWatchInitConnBackoff(t *testing.T) {
store: NewFIFO(MetaNamespaceKeyFunc),
backoffManager: bm,
clock: fakeClock,
watchErrorHandler: WatchErrorHandler(DefaultWatchErrorHandler),
watchErrorHandler: WatchErrorHandlerWithContext(DefaultWatchErrorHandler),
}
start := fakeClock.Now()
err := r.ListAndWatch(stopCh)
err := r.ListAndWatchWithContext(ctx)
elapsed := fakeClock.Since(start)
if err != nil {
t.Errorf("unexpected error %v", err)
@@ -666,6 +682,7 @@ func (f *fakeBackoff) Backoff() clock.Timer {
}
func TestBackoffOnTooManyRequests(t *testing.T) {
_, ctx := ktesting.NewTestContext(t)
err := apierrors.NewTooManyRequests("too many requests", 1)
clock := &clock.RealClock{}
bm := &fakeBackoff{clock: clock}
@@ -697,11 +714,11 @@ func TestBackoffOnTooManyRequests(t *testing.T) {
store: NewFIFO(MetaNamespaceKeyFunc),
backoffManager: bm,
clock: clock,
watchErrorHandler: WatchErrorHandler(DefaultWatchErrorHandler),
watchErrorHandler: WatchErrorHandlerWithContext(DefaultWatchErrorHandler),
}
stopCh := make(chan struct{})
if err := r.ListAndWatch(stopCh); err != nil {
if err := r.ListAndWatchWithContext(ctx); err != nil {
t.Fatal(err)
}
close(stopCh)
@@ -738,14 +755,15 @@ func TestNoRelistOnTooManyRequests(t *testing.T) {
store: NewFIFO(MetaNamespaceKeyFunc),
backoffManager: bm,
clock: clock,
watchErrorHandler: WatchErrorHandler(DefaultWatchErrorHandler),
watchErrorHandler: WatchErrorHandlerWithContext(DefaultWatchErrorHandler),
}
stopCh := make(chan struct{})
if err := r.ListAndWatch(stopCh); err != nil {
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancelCause(ctx)
if err := r.ListAndWatchWithContext(ctx); err != nil {
t.Fatal(err)
}
close(stopCh)
cancel(errors.New("done"))
if listCalls != 1 {
t.Errorf("unexpected list calls: %d", listCalls)
}
@@ -812,14 +830,15 @@ func TestRetryInternalError(t *testing.T) {
store: NewFIFO(MetaNamespaceKeyFunc),
backoffManager: bm,
clock: fakeClock,
watchErrorHandler: WatchErrorHandler(DefaultWatchErrorHandler),
watchErrorHandler: WatchErrorHandlerWithContext(DefaultWatchErrorHandler),
}
r.MaxInternalErrorRetryDuration = tc.maxInternalDuration
stopCh := make(chan struct{})
r.ListAndWatch(stopCh)
close(stopCh)
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancelCause(ctx)
require.NoError(t, r.ListAndWatchWithContext(ctx))
cancel(errors.New("done"))
if counter-1 != tc.wantRetries {
t.Errorf("%v unexpected number of retries: %d", tc, counter-1)
@@ -829,7 +848,7 @@ func TestRetryInternalError(t *testing.T) {
func TestReflectorResync(t *testing.T) {
iteration := 0
stopCh := make(chan struct{})
_, ctx := ktesting.NewTestContext(t)
rerr := errors.New("expected resync reached")
s := &FakeCustomStore{
ResyncFunc: func() error {
@@ -852,7 +871,7 @@ func TestReflectorResync(t *testing.T) {
}
resyncPeriod := 1 * time.Millisecond
r := NewReflector(lw, &v1.Pod{}, s, resyncPeriod)
if err := r.ListAndWatch(stopCh); err != nil {
if err := r.ListAndWatchWithContext(ctx); err != nil {
// error from Resync is not propaged up to here.
t.Errorf("expected error %v", err)
}
@@ -862,13 +881,14 @@ func TestReflectorResync(t *testing.T) {
}
func TestReflectorWatchListPageSize(t *testing.T) {
stopCh := make(chan struct{})
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancelCause(ctx)
s := NewStore(MetaNamespaceKeyFunc)
lw := &testLW{
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
// Stop once the reflector begins watching since we're only interested in the list.
close(stopCh)
cancel(errors.New("done"))
fw := watch.NewFake()
return fw, nil
},
@@ -898,7 +918,7 @@ func TestReflectorWatchListPageSize(t *testing.T) {
r.setLastSyncResourceVersion("10")
// Set the reflector to paginate the list request in 4 item chunks.
r.WatchListPageSize = 4
r.ListAndWatch(stopCh)
require.NoError(t, r.ListAndWatchWithContext(ctx))
results := s.List()
if len(results) != 10 {
@@ -907,13 +927,14 @@ func TestReflectorWatchListPageSize(t *testing.T) {
}
func TestReflectorNotPaginatingNotConsistentReads(t *testing.T) {
stopCh := make(chan struct{})
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancelCause(ctx)
s := NewStore(MetaNamespaceKeyFunc)
lw := &testLW{
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
// Stop once the reflector begins watching since we're only interested in the list.
close(stopCh)
cancel(errors.New("done"))
fw := watch.NewFake()
return fw, nil
},
@@ -933,7 +954,7 @@ func TestReflectorNotPaginatingNotConsistentReads(t *testing.T) {
}
r := NewReflector(lw, &v1.Pod{}, s, 0)
r.setLastSyncResourceVersion("10")
r.ListAndWatch(stopCh)
require.NoError(t, r.ListAndWatchWithContext(ctx))
results := s.List()
if len(results) != 10 {
@@ -942,13 +963,14 @@ func TestReflectorNotPaginatingNotConsistentReads(t *testing.T) {
}
func TestReflectorPaginatingNonConsistentReadsIfWatchCacheDisabled(t *testing.T) {
var stopCh chan struct{}
_, ctx := ktesting.NewTestContext(t)
var cancel func(error)
s := NewStore(MetaNamespaceKeyFunc)
lw := &testLW{
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
// Stop once the reflector begins watching since we're only interested in the list.
close(stopCh)
cancel(errors.New("done"))
fw := watch.NewFake()
return fw, nil
},
@@ -977,16 +999,17 @@ func TestReflectorPaginatingNonConsistentReadsIfWatchCacheDisabled(t *testing.T)
r := NewReflector(lw, &v1.Pod{}, s, 0)
// Initial list should initialize paginatedResult in the reflector.
stopCh = make(chan struct{})
r.ListAndWatch(stopCh)
var cancelCtx context.Context
cancelCtx, cancel = context.WithCancelCause(ctx)
require.NoError(t, r.ListAndWatchWithContext(cancelCtx))
if results := s.List(); len(results) != 10 {
t.Errorf("Expected 10 results, got %d", len(results))
}
// Since initial list for ResourceVersion="0" was paginated, the subsequent
// ones should also be paginated.
stopCh = make(chan struct{})
r.ListAndWatch(stopCh)
cancelCtx, cancel = context.WithCancelCause(ctx)
require.NoError(t, r.ListAndWatchWithContext(cancelCtx))
if results := s.List(); len(results) != 10 {
t.Errorf("Expected 10 results, got %d", len(results))
}
@@ -996,14 +1019,15 @@ func TestReflectorPaginatingNonConsistentReadsIfWatchCacheDisabled(t *testing.T)
// it in relist requests to prevent the reflector from traveling back in time if the relist is to a api-server or
// etcd that is partitioned and serving older data than the reflector has already processed.
func TestReflectorResyncWithResourceVersion(t *testing.T) {
stopCh := make(chan struct{})
_, ctx := ktesting.NewTestContext(t)
cancelCtx, cancel := context.WithCancelCause(ctx)
s := NewStore(MetaNamespaceKeyFunc)
listCallRVs := []string{}
lw := &testLW{
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
// Stop once the reflector begins watching since we're only interested in the list.
close(stopCh)
cancel(errors.New("done"))
fw := watch.NewFake()
return fw, nil
},
@@ -1027,7 +1051,7 @@ func TestReflectorResyncWithResourceVersion(t *testing.T) {
r := NewReflector(lw, &v1.Pod{}, s, 0)
// Initial list should use RV=0
r.ListAndWatch(stopCh)
require.NoError(t, r.ListAndWatchWithContext(cancelCtx))
results := s.List()
if len(results) != 4 {
@@ -1035,8 +1059,8 @@ func TestReflectorResyncWithResourceVersion(t *testing.T) {
}
// relist should use lastSyncResourceVersions (RV=10)
stopCh = make(chan struct{})
r.ListAndWatch(stopCh)
cancelCtx, cancel = context.WithCancelCause(ctx)
require.NoError(t, r.ListAndWatchWithContext(cancelCtx))
results = s.List()
if len(results) != 8 {
@@ -1055,14 +1079,16 @@ func TestReflectorResyncWithResourceVersion(t *testing.T) {
// (In kubernetes 1.17, or when the watch cache is enabled, the List will instead return the list that is no older than
// the requested ResourceVersion).
func TestReflectorExpiredExactResourceVersion(t *testing.T) {
stopCh := make(chan struct{})
_, ctx := ktesting.NewTestContext(t)
var cancelCtx context.Context
var cancel func(error)
s := NewStore(MetaNamespaceKeyFunc)
listCallRVs := []string{}
lw := &testLW{
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
// Stop once the reflector begins watching since we're only interested in the list.
close(stopCh)
cancel(errors.New("done"))
fw := watch.NewFake()
return fw, nil
},
@@ -1089,7 +1115,8 @@ func TestReflectorExpiredExactResourceVersion(t *testing.T) {
r := NewReflector(lw, &v1.Pod{}, s, 0)
// Initial list should use RV=0
r.ListAndWatch(stopCh)
cancelCtx, cancel = context.WithCancelCause(ctx)
require.NoError(t, r.ListAndWatchWithContext(cancelCtx))
results := s.List()
if len(results) != 4 {
@@ -1097,8 +1124,8 @@ func TestReflectorExpiredExactResourceVersion(t *testing.T) {
}
// relist should use lastSyncResourceVersions (RV=10) and since RV=10 is expired, it should retry with RV="".
stopCh = make(chan struct{})
r.ListAndWatch(stopCh)
cancelCtx, cancel = context.WithCancelCause(ctx)
require.NoError(t, r.ListAndWatchWithContext(cancelCtx))
results = s.List()
if len(results) != 8 {
@@ -1112,14 +1139,16 @@ func TestReflectorExpiredExactResourceVersion(t *testing.T) {
}
func TestReflectorFullListIfExpired(t *testing.T) {
stopCh := make(chan struct{})
_, ctx := ktesting.NewTestContext(t)
var cancelCtx context.Context
var cancel func(error)
s := NewStore(MetaNamespaceKeyFunc)
listCallRVs := []string{}
lw := &testLW{
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
// Stop once the reflector begins watching since we're only interested in the list.
close(stopCh)
cancel(errors.New("done"))
fw := watch.NewFake()
return fw, nil
},
@@ -1156,7 +1185,8 @@ func TestReflectorFullListIfExpired(t *testing.T) {
r.WatchListPageSize = 4
// Initial list should use RV=0
if err := r.ListAndWatch(stopCh); err != nil {
cancelCtx, cancel = context.WithCancelCause(ctx)
if err := r.ListAndWatchWithContext(cancelCtx); err != nil {
t.Fatal(err)
}
@@ -1166,8 +1196,8 @@ func TestReflectorFullListIfExpired(t *testing.T) {
}
// relist should use lastSyncResourceVersions (RV=10) and since second page of that expired, it should full list with RV=10
stopCh = make(chan struct{})
if err := r.ListAndWatch(stopCh); err != nil {
cancelCtx, cancel = context.WithCancelCause(ctx)
if err := r.ListAndWatchWithContext(cancelCtx); err != nil {
t.Fatal(err)
}
@@ -1183,7 +1213,9 @@ func TestReflectorFullListIfExpired(t *testing.T) {
}
func TestReflectorFullListIfTooLarge(t *testing.T) {
stopCh := make(chan struct{})
_, ctx := ktesting.NewTestContext(t)
var cancelCtx context.Context
var cancel func(error)
s := NewStore(MetaNamespaceKeyFunc)
listCallRVs := []string{}
version := 30
@@ -1191,7 +1223,7 @@ func TestReflectorFullListIfTooLarge(t *testing.T) {
lw := &testLW{
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
// Stop once the reflector begins watching since we're only interested in the list.
close(stopCh)
cancel(errors.New("done"))
fw := watch.NewFake()
return fw, nil
},
@@ -1229,7 +1261,8 @@ func TestReflectorFullListIfTooLarge(t *testing.T) {
r := NewReflector(lw, &v1.Pod{}, s, 0)
// Initial list should use RV=0
if err := r.ListAndWatch(stopCh); err != nil {
cancelCtx, cancel = context.WithCancelCause(ctx)
if err := r.ListAndWatchWithContext(cancelCtx); err != nil {
t.Fatal(err)
}
@@ -1241,8 +1274,8 @@ func TestReflectorFullListIfTooLarge(t *testing.T) {
// done we simply try to relist from now to avoid continuous errors on relists.
for i := 1; i <= 3; i++ {
// relist twice to cover the two variants of TooLargeResourceVersion api errors
stopCh = make(chan struct{})
if err := r.ListAndWatch(stopCh); err != nil {
cancelCtx, cancel = context.WithCancelCause(ctx)
if err := r.ListAndWatchWithContext(cancelCtx); err != nil {
t.Fatal(err)
}
}
@@ -1356,7 +1389,8 @@ func TestWatchTimeout(t *testing.T) {
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
stopCh := make(chan struct{})
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancelCause(ctx)
s := NewStore(MetaNamespaceKeyFunc)
var gotTimeoutSeconds int64
@@ -1370,7 +1404,7 @@ func TestWatchTimeout(t *testing.T) {
}
// Stop once the reflector begins watching since we're only interested in the list.
close(stopCh)
cancel(errors.New("done"))
return watch.NewFake(), nil
},
}
@@ -1379,7 +1413,7 @@ func TestWatchTimeout(t *testing.T) {
MinWatchTimeout: tc.minWatchTimeout,
}
r := NewReflectorWithOptions(lw, &v1.Pod{}, s, opts)
if err := r.ListAndWatch(stopCh); err != nil {
if err := r.ListAndWatchWithContext(ctx); err != nil {
t.Fatal(err)
}
@@ -1412,7 +1446,8 @@ func newStoreWithRV() *storeWithRV {
func TestReflectorResourceVersionUpdate(t *testing.T) {
s := newStoreWithRV()
stopCh := make(chan struct{})
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancelCause(ctx)
fw := watch.NewFake()
lw := &testLW{
@@ -1434,11 +1469,11 @@ func TestReflectorResourceVersionUpdate(t *testing.T) {
fw.Action(watch.Modified, makePod("20"))
fw.Action(watch.Bookmark, makePod("30"))
fw.Action(watch.Deleted, makePod("40"))
close(stopCh)
cancel(errors.New("done"))
}()
// Initial list should use RV=0
if err := r.ListAndWatch(stopCh); err != nil {
if err := r.ListAndWatchWithContext(ctx); err != nil {
t.Fatal(err)
}
@@ -1680,6 +1715,7 @@ func (t *TestPagingPodsLW) Watch(options metav1.ListOptions) (watch.Interface, e
}
func TestReflectorListExtract(t *testing.T) {
_, ctx := ktesting.NewTestContext(t)
store := NewStore(func(obj interface{}) (string, error) {
pod, ok := obj.(*v1.Pod)
if !ok {
@@ -1693,8 +1729,7 @@ func TestReflectorListExtract(t *testing.T) {
reflector.WatchListPageSize = fakeItemsNum
// execute list to fill store
stopCh := make(chan struct{})
if err := reflector.list(stopCh); err != nil {
if err := reflector.list(ctx); err != nil {
t.Fatal(err)
}
@@ -2074,7 +2109,7 @@ func BenchmarkReflectorList(b *testing.B) {
b.ResetTimer()
for i := 0; i < b.N; i++ {
err := reflector.list(ctx.Done())
err := reflector.list(ctx)
if err != nil {
b.Fatalf("reflect list: %v", err)
}

23
staging/src/k8s.io/client-go/tools/cache/reflector_watchlist_test.go vendored
View File

@@ -17,6 +17,7 @@ limitations under the License.
package cache
import (
"context"
"errors"
"fmt"
"sort"
@@ -35,6 +36,7 @@ import (
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/klog/v2/ktesting"
testingclock "k8s.io/utils/clock/testing"
"k8s.io/utils/pointer"
"k8s.io/utils/ptr"
@@ -51,8 +53,9 @@ func TestInitialEventsEndBookmarkTicker(t *testing.T) {
}
t.Run("testing NoopInitialEventsEndBookmarkTicker", func(t *testing.T) {
logger, _ := ktesting.NewTestContext(t)
clock := testingclock.NewFakeClock(time.Now())
target := newInitialEventsEndBookmarkTickerInternal("testName", clock, clock.Now(), time.Second, false)
target := newInitialEventsEndBookmarkTickerInternal(logger, "testName", clock, clock.Now(), time.Second, false)
clock.Step(30 * time.Second)
assertNoEvents(t, target.C())
@@ -73,8 +76,9 @@ func TestInitialEventsEndBookmarkTicker(t *testing.T) {
})
t.Run("testing InitialEventsEndBookmarkTicker backed by a fake clock", func(t *testing.T) {
logger, _ := ktesting.NewTestContext(t)
clock := testingclock.NewFakeClock(time.Now())
target := newInitialEventsEndBookmarkTickerInternal("testName", clock, clock.Now(), time.Second, true)
target := newInitialEventsEndBookmarkTickerInternal(logger, "testName", clock, clock.Now(), time.Second, true)
clock.Step(500 * time.Millisecond)
assertNoEvents(t, target.C())
@@ -469,7 +473,8 @@ func TestWatchList(t *testing.T) {
for _, s := range scenarios {
t.Run(s.name, func(t *testing.T) {
scenario := s // capture as local variable
listWatcher, store, reflector, stopCh := testData()
_, ctx := ktesting.NewTestContext(t)
listWatcher, store, reflector, ctx, cancel := testData(ctx)
go func() {
for i, e := range scenario.watchEvents {
listWatcher.fakeWatcher.Action(e.Type, e.Object)
@@ -478,7 +483,7 @@ func TestWatchList(t *testing.T) {
continue
}
if i+1 == scenario.closeAfterWatchEvents {
close(stopCh)
cancel(fmt.Errorf("done after %d watch events", i))
}
}
}()
@@ -490,7 +495,7 @@ func TestWatchList(t *testing.T) {
reflector.UseWatchList = ptr.To(false)
}
err := reflector.ListAndWatch(stopCh)
err := reflector.ListAndWatchWithContext(ctx)
if scenario.expectedError != nil && err == nil {
t.Fatalf("expected error %q, got nil", scenario.expectedError)
}
@@ -567,19 +572,19 @@ func makePod(name, rv string) *v1.Pod {
return &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: name, ResourceVersion: rv, UID: types.UID(name)}}
}
func testData() (*fakeListWatcher, Store, *Reflector, chan struct{}) {
func testData(ctx context.Context) (*fakeListWatcher, Store, *Reflector, context.Context, func(error)) {
ctx, cancel := context.WithCancelCause(ctx)
s := NewStore(MetaNamespaceKeyFunc)
stopCh := make(chan struct{})
lw := &fakeListWatcher{
fakeWatcher: watch.NewFake(),
stop: func() {
close(stopCh)
cancel(errors.New("time to stop"))
},
}
r := NewReflector(lw, &v1.Pod{}, s, 0)
r.UseWatchList = ptr.To(true)
return lw, s, r, stopCh
return lw, s, r, ctx, cancel
}
type fakeListWatcher struct {

160
staging/src/k8s.io/client-go/tools/cache/shared_informer.go vendored
View File

@@ -17,6 +17,7 @@ limitations under the License.
package cache
import (
"context"
"errors"
"fmt"
"sync"
@@ -29,6 +30,7 @@ import (
"k8s.io/client-go/tools/cache/synctrack"
"k8s.io/utils/buffer"
"k8s.io/utils/clock"
"k8s.io/utils/ptr"
"k8s.io/klog/v2"
@@ -142,6 +144,8 @@ type SharedInformer interface {
// It returns a registration handle for the handler that can be used to
// remove the handler again, or to tell if the handler is synced (has
// seen every item in the initial list).
//
// Contextual logging: AddEventHandlerWithOptions together with a logger in the options should be used instead of AddEventHandler in code which supports contextual logging.
AddEventHandler(handler ResourceEventHandler) (ResourceEventHandlerRegistration, error)
// AddEventHandlerWithResyncPeriod adds an event handler to the
// shared informer with the requested resync period; zero means
@@ -159,7 +163,12 @@ type SharedInformer interface {
// be competing load and scheduling noise.
// It returns a registration handle for the handler that can be used to remove
// the handler again and an error if the handler cannot be added.
//
// Contextual logging: AddEventHandlerWithOptions together with a logger in the options should be used instead of AddEventHandlerWithResyncPeriod in code which supports contextual logging.
AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration) (ResourceEventHandlerRegistration, error)
// AddEventHandlerWithOptions is a variant of AddEventHandlerWithResyncPeriod where
// all optional parameters are passed in a struct.
AddEventHandlerWithOptions(handler ResourceEventHandler, options HandlerOptions) (ResourceEventHandlerRegistration, error)
// RemoveEventHandler removes a formerly added event handler given by
// its registration handle.
// This function is guaranteed to be idempotent, and thread-safe.
@@ -170,7 +179,12 @@ type SharedInformer interface {
GetController() Controller
// Run starts and runs the shared informer, returning after it stops.
// The informer will be stopped when stopCh is closed.
//
// Contextual logging: RunWithContext should be used instead of Run in code which uses contextual logging.
Run(stopCh <-chan struct{})
// RunWithContext starts and runs the shared informer, returning after it stops.
// The informer will be stopped when the context is canceled.
RunWithContext(ctx context.Context)
// HasSynced returns true if the shared informer's store has been
// informed by at least one full LIST of the authoritative state
// of the informer's object collection. This is unrelated to "resync".
@@ -197,8 +211,14 @@ type SharedInformer interface {
// The handler is intended for visibility, not to e.g. pause the consumers.
// The handler should return quickly - any expensive processing should be
// offloaded.
//
// Contextual logging: SetWatchErrorHandlerWithContext should be used instead of SetWatchErrorHandler in code which supports contextual logging.
SetWatchErrorHandler(handler WatchErrorHandler) error
// SetWatchErrorHandlerWithContext is a variant of SetWatchErrorHandler where
// the handler is passed an additional context parameter.
SetWatchErrorHandlerWithContext(handler WatchErrorHandlerWithContext) error
// The TransformFunc is called for each object which is about to be stored.
//
// This function is intended for you to take the opportunity to
@@ -228,6 +248,21 @@ type ResourceEventHandlerRegistration interface {
HasSynced() bool
}
// Optional configuration options for [SharedInformer.AddEventHandlerWithOptions].
// May be left empty.
type HandlerOptions struct {
// Logger overrides the default klog.Background() logger.
Logger *klog.Logger
// ResyncPeriod requests a certain resync period from an informer. Zero
// means the handler does not care about resyncs. Not all informers do
// resyncs, even if requested. See
// [SharedInformer.AddEventHandlerWithResyncPeriod] for details.
//
// If nil, the default resync period of the shared informer is used.
ResyncPeriod *time.Duration
}
// SharedIndexInformer provides add and get Indexers ability based on SharedInformer.
type SharedIndexInformer interface {
SharedInformer
@@ -309,15 +344,38 @@ const (
// WaitForNamedCacheSync is a wrapper around WaitForCacheSync that generates log messages
// indicating that the caller identified by name is waiting for syncs, followed by
// either a successful or failed sync.
//
// Contextual logging: WaitForNamedCacheSyncWithContext should be used instead of WaitForNamedCacheSync in code which supports contextual logging.
func WaitForNamedCacheSync(controllerName string, stopCh <-chan struct{}, cacheSyncs ...InformerSynced) bool {
klog.Infof("Waiting for caches to sync for %s", controllerName)
klog.Background().Info("Waiting for caches to sync", "controller", controllerName)
if !WaitForCacheSync(stopCh, cacheSyncs...) {
utilruntime.HandleError(fmt.Errorf("unable to sync caches for %s", controllerName))
utilruntime.HandleErrorWithContext(context.Background(), nil, "Unable to sync caches", "controller", controllerName)
return false
}
klog.Infof("Caches are synced for %s", controllerName)
klog.Background().Info("Caches are synced", "controller", controllerName)
return true
}
// WaitForNamedCacheSyncWithContext is a wrapper around WaitForCacheSyncWithContext that generates log messages
// indicating that the caller is waiting for syncs, followed by either a successful or failed sync.
//
// Contextual logging can be used to identify the caller in those log messages. The log level is zero,
// the same as in [WaitForNamedCacheSync]. If this is too verbose, then store a logger with an increased
// threshold in the context:
//
// WaitForNamedCacheSyncWithContext(klog.NewContext(ctx, logger.V(5)), ...)
func WaitForNamedCacheSyncWithContext(ctx context.Context, cacheSyncs ...InformerSynced) bool {
logger := klog.FromContext(ctx)
logger.Info("Waiting for caches to sync")
if !WaitForCacheSync(ctx.Done(), cacheSyncs...) {
utilruntime.HandleErrorWithContext(ctx, nil, "Unable to sync caches")
return false
}
logger.Info("Caches are synced")
return true
}
@@ -389,7 +447,7 @@ type sharedIndexInformer struct {
blockDeltas sync.Mutex
// Called whenever the ListAndWatch drops the connection with an error.
watchErrorHandler WatchErrorHandler
watchErrorHandler WatchErrorHandlerWithContext
transform TransformFunc
}
@@ -403,6 +461,9 @@ type dummyController struct {
informer *sharedIndexInformer
}
func (v *dummyController) RunWithContext(context.Context) {
}
func (v *dummyController) Run(stopCh <-chan struct{}) {
}
@@ -433,6 +494,12 @@ type deleteNotification struct {
}
func (s *sharedIndexInformer) SetWatchErrorHandler(handler WatchErrorHandler) error {
return s.SetWatchErrorHandlerWithContext(func(_ context.Context, r *Reflector, err error) {
handler(r, err)
})
}
func (s *sharedIndexInformer) SetWatchErrorHandlerWithContext(handler WatchErrorHandlerWithContext) error {
s.startedLock.Lock()
defer s.startedLock.Unlock()
@@ -457,10 +524,15 @@ func (s *sharedIndexInformer) SetTransform(handler TransformFunc) error {
}
func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
defer utilruntime.HandleCrash()
s.RunWithContext(wait.ContextForChannel(stopCh))
}
func (s *sharedIndexInformer) RunWithContext(ctx context.Context) {
defer utilruntime.HandleCrashWithContext(ctx)
logger := klog.FromContext(ctx)
if s.HasStarted() {
klog.Warningf("The sharedIndexInformer has started, run more than once is not allowed")
logger.Info("Warning: the sharedIndexInformer has started, run more than once is not allowed")
return
}
@@ -484,7 +556,7 @@ func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
ShouldResync: s.processor.shouldResync,
Process: s.HandleDeltas,
WatchErrorHandler: s.watchErrorHandler,
WatchErrorHandlerWithContext: s.watchErrorHandler,
}
s.controller = New(cfg)
@@ -492,20 +564,24 @@ func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
s.started = true
}()
// Separate stop channel because Processor should be stopped strictly after controller
processorStopCh := make(chan struct{})
// Separate stop context because Processor should be stopped strictly after controller.
// Cancelation in the parent context is ignored and all values are passed on,
// including - but not limited to - a logger.
processorStopCtx, stopProcessor := context.WithCancelCause(context.WithoutCancel(ctx))
var wg wait.Group
defer wg.Wait() // Wait for Processor to stop
defer close(processorStopCh) // Tell Processor to stop
wg.StartWithChannel(processorStopCh, s.cacheMutationDetector.Run)
wg.StartWithChannel(processorStopCh, s.processor.run)
defer stopProcessor(errors.New("informer is stopping")) // Tell Processor to stop
// TODO: extend the MutationDetector interface so that it optionally
// has a RunWithContext method that we can use here.
wg.StartWithChannel(processorStopCtx.Done(), s.cacheMutationDetector.Run)
wg.StartWithContext(processorStopCtx, s.processor.run)
defer func() {
s.startedLock.Lock()
defer s.startedLock.Unlock()
s.stopped = true // Don't want any new listeners
}()
s.controller.Run(stopCh)
s.controller.RunWithContext(ctx)
}
func (s *sharedIndexInformer) HasStarted() bool {
@@ -558,19 +634,19 @@ func (s *sharedIndexInformer) GetController() Controller {
}
func (s *sharedIndexInformer) AddEventHandler(handler ResourceEventHandler) (ResourceEventHandlerRegistration, error) {
return s.AddEventHandlerWithResyncPeriod(handler, s.defaultEventHandlerResyncPeriod)
return s.AddEventHandlerWithOptions(handler, HandlerOptions{})
}
func determineResyncPeriod(desired, check time.Duration) time.Duration {
func determineResyncPeriod(logger klog.Logger, desired, check time.Duration) time.Duration {
if desired == 0 {
return desired
}
if check == 0 {
klog.Warningf("The specified resyncPeriod %v is invalid because this shared informer doesn't support resyncing", desired)
logger.Info("Warning: the specified resyncPeriod is invalid because this shared informer doesn't support resyncing", "desired", desired)
return 0
}
if desired < check {
klog.Warningf("The specified resyncPeriod %v is being increased to the minimum resyncCheckPeriod %v", desired, check)
logger.Info("Warning: the specified resyncPeriod is being increased to the minimum resyncCheckPeriod", "desired", desired, "resyncCheckPeriod", check)
return check
}
return desired
@@ -579,6 +655,10 @@ func determineResyncPeriod(desired, check time.Duration) time.Duration {
const minimumResyncPeriod = 1 * time.Second
func (s *sharedIndexInformer) AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration) (ResourceEventHandlerRegistration, error) {
return s.AddEventHandlerWithOptions(handler, HandlerOptions{ResyncPeriod: &resyncPeriod})
}
func (s *sharedIndexInformer) AddEventHandlerWithOptions(handler ResourceEventHandler, options HandlerOptions) (ResourceEventHandlerRegistration, error) {
s.startedLock.Lock()
defer s.startedLock.Unlock()
@@ -586,27 +666,30 @@ func (s *sharedIndexInformer) AddEventHandlerWithResyncPeriod(handler ResourceEv
return nil, fmt.Errorf("handler %v was not added to shared informer because it has stopped already", handler)
}
logger := ptr.Deref(options.Logger, klog.Background())
resyncPeriod := ptr.Deref(options.ResyncPeriod, s.defaultEventHandlerResyncPeriod)
if resyncPeriod > 0 {
if resyncPeriod < minimumResyncPeriod {
klog.Warningf("resyncPeriod %v is too small. Changing it to the minimum allowed value of %v", resyncPeriod, minimumResyncPeriod)
logger.Info("Warning: resync period is too small. Changing it to the minimum allowed value", "resyncPeriod", resyncPeriod, "minimumResyncPeriod", minimumResyncPeriod)
resyncPeriod = minimumResyncPeriod
}
if resyncPeriod < s.resyncCheckPeriod {
if s.started {
klog.Warningf("resyncPeriod %v is smaller than resyncCheckPeriod %v and the informer has already started. Changing it to %v", resyncPeriod, s.resyncCheckPeriod, s.resyncCheckPeriod)
logger.Info("Warning: resync period is smaller than resync check period and the informer has already started. Changing it to the resync check period", "resyncPeriod", resyncPeriod, "resyncCheckPeriod", s.resyncCheckPeriod)
resyncPeriod = s.resyncCheckPeriod
} else {
// if the event handler's resyncPeriod is smaller than the current resyncCheckPeriod, update
// resyncCheckPeriod to match resyncPeriod and adjust the resync periods of all the listeners
// accordingly
s.resyncCheckPeriod = resyncPeriod
s.processor.resyncCheckPeriodChanged(resyncPeriod)
s.processor.resyncCheckPeriodChanged(logger, resyncPeriod)
}
}
}
listener := newProcessListener(handler, resyncPeriod, determineResyncPeriod(resyncPeriod, s.resyncCheckPeriod), s.clock.Now(), initialBufferSize, s.HasSynced)
listener := newProcessListener(logger, handler, resyncPeriod, determineResyncPeriod(logger, resyncPeriod, s.resyncCheckPeriod), s.clock.Now(), initialBufferSize, s.HasSynced)
if !s.started {
return s.processor.addListener(listener), nil
@@ -794,7 +877,7 @@ func (p *sharedProcessor) distribute(obj interface{}, sync bool) {
}
}
func (p *sharedProcessor) run(stopCh <-chan struct{}) {
func (p *sharedProcessor) run(ctx context.Context) {
func() {
p.listenersLock.RLock()
defer p.listenersLock.RUnlock()
@@ -804,7 +887,7 @@ func (p *sharedProcessor) run(stopCh <-chan struct{}) {
}
p.listenersStarted = true
}()
<-stopCh
<-ctx.Done()
p.listenersLock.Lock()
defer p.listenersLock.Unlock()
@@ -844,13 +927,13 @@ func (p *sharedProcessor) shouldResync() bool {
return resyncNeeded
}
func (p *sharedProcessor) resyncCheckPeriodChanged(resyncCheckPeriod time.Duration) {
func (p *sharedProcessor) resyncCheckPeriodChanged(logger klog.Logger, resyncCheckPeriod time.Duration) {
p.listenersLock.RLock()
defer p.listenersLock.RUnlock()
for listener := range p.listeners {
resyncPeriod := determineResyncPeriod(
listener.requestedResyncPeriod, resyncCheckPeriod)
logger, listener.requestedResyncPeriod, resyncCheckPeriod)
listener.setResyncPeriod(resyncPeriod)
}
}
@@ -867,6 +950,7 @@ func (p *sharedProcessor) resyncCheckPeriodChanged(resyncCheckPeriod time.Durati
// processorListener also keeps track of the adjusted requested resync
// period of the listener.
type processorListener struct {
logger klog.Logger
nextCh chan interface{}
addCh chan interface{}
@@ -910,8 +994,9 @@ func (p *processorListener) HasSynced() bool {
return p.syncTracker.HasSynced()
}
func newProcessListener(handler ResourceEventHandler, requestedResyncPeriod, resyncPeriod time.Duration, now time.Time, bufferSize int, hasSynced func() bool) *processorListener {
func newProcessListener(logger klog.Logger, handler ResourceEventHandler, requestedResyncPeriod, resyncPeriod time.Duration, now time.Time, bufferSize int, hasSynced func() bool) *processorListener {
ret := &processorListener{
logger: logger,
nextCh: make(chan interface{}),
addCh: make(chan interface{}),
handler: handler,
@@ -934,7 +1019,7 @@ func (p *processorListener) add(notification interface{}) {
}
func (p *processorListener) pop() {
defer utilruntime.HandleCrash()
defer utilruntime.HandleCrashWithLogger(p.logger)
defer close(p.nextCh) // Tell .run() to stop
var nextCh chan<- interface{}
@@ -968,9 +1053,19 @@ func (p *processorListener) run() {
// we will catch it, **the offending item will be skipped!**, and after a short delay (one second)
// the next notification will be attempted. This is usually better than the alternative of never
// delivering again.
stopCh := make(chan struct{})
wait.Until(func() {
//
// This only applies if utilruntime is configured to not panic, which is not the default.
sleepAfterCrash := false
for next := range p.nextCh {
if sleepAfterCrash {
// Sleep before processing the next item.
time.Sleep(time.Second)
}
func() {
// Gets reset below, but only if we get that far.
sleepAfterCrash = true
defer utilruntime.HandleCrashWithLogger(p.logger)
switch notification := next.(type) {
case updateNotification:
p.handler.OnUpdate(notification.oldObj, notification.newObj)
@@ -982,12 +1077,11 @@ func (p *processorListener) run() {
case deleteNotification:
p.handler.OnDelete(notification.oldObj)
default:
utilruntime.HandleError(fmt.Errorf("unrecognized notification: %T", next))
utilruntime.HandleErrorWithLogger(p.logger, nil, "unrecognized notification", "notificationType", fmt.Sprintf("%T", next))
}
sleepAfterCrash = false
}()
}
// the only way to get here is if the p.nextCh is empty and closed
close(stopCh)
}, 1*time.Second, stopCh)
}
// shouldResync deterimines if the listener needs a resync. If the listener's resyncPeriod is 0,

347
staging/src/k8s.io/client-go/tools/cache/shared_informer_test.go vendored
View File

@@ -20,6 +20,7 @@ import (
"context"
"fmt"
"math/rand"
"runtime"
"strconv"
"strings"
"sync"
@@ -29,12 +30,18 @@ import (
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apimachinery/pkg/util/wait"
fcache "k8s.io/client-go/tools/cache/testing"
"k8s.io/klog/v2"
"k8s.io/klog/v2/ktesting"
"k8s.io/klog/v2/textlogger"
testingclock "k8s.io/utils/clock/testing"
)
@@ -123,7 +130,7 @@ func isRegistered(i SharedInformer, h ResourceEventHandlerRegistration) bool {
func TestIndexer(t *testing.T) {
assert := assert.New(t)
// source simulates an apiserver object endpoint.
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
pod1 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1", Labels: map[string]string{"a": "a-val", "b": "b-val1"}}}
pod2 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod2", Labels: map[string]string{"b": "b-val2"}}}
pod3 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod3", Labels: map[string]string{"a": "a-val2"}}}
@@ -144,10 +151,15 @@ func TestIndexer(t *testing.T) {
if err != nil {
t.Fatal(err)
}
stop := make(chan struct{})
defer close(stop)
go informer.Run(stop)
var wg wait.Group
stop := make(chan struct{})
wg.StartWithChannel(stop, informer.Run)
defer func() {
close(stop)
wg.Wait()
}()
WaitForCacheSync(stop, informer.HasSynced)
cmpOps := cmpopts.SortSlices(func(a, b any) bool {
@@ -197,7 +209,7 @@ func TestIndexer(t *testing.T) {
func TestListenerResyncPeriods(t *testing.T) {
// source simulates an apiserver object endpoint.
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod2"}})
@@ -221,10 +233,13 @@ func TestListenerResyncPeriods(t *testing.T) {
informer.AddEventHandlerWithResyncPeriod(listener3, listener3.resyncPeriod)
listeners := []*testListener{listener1, listener2, listener3}
var wg wait.Group
stop := make(chan struct{})
defer close(stop)
go informer.Run(stop)
wg.StartWithChannel(stop, informer.Run)
defer func() {
close(stop)
wg.Wait()
}()
// ensure all listeners got the initial List
for _, listener := range listeners {
@@ -284,7 +299,7 @@ func TestListenerResyncPeriods(t *testing.T) {
func TestResyncCheckPeriod(t *testing.T) {
// source simulates an apiserver object endpoint.
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
// create the shared informer and resync every 12 hours
informer := NewSharedInformer(source, &v1.Pod{}, 12*time.Hour).(*sharedIndexInformer)
@@ -356,14 +371,18 @@ func TestResyncCheckPeriod(t *testing.T) {
// verify that https://github.com/kubernetes/kubernetes/issues/59822 is fixed
func TestSharedInformerInitializationRace(t *testing.T) {
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second).(*sharedIndexInformer)
listener := newTestListener("raceListener", 0)
stop := make(chan struct{})
go informer.AddEventHandlerWithResyncPeriod(listener, listener.resyncPeriod)
go informer.Run(stop)
var wg wait.Group
stop := make(chan struct{})
wg.StartWithChannel(stop, informer.Run)
defer func() {
close(stop)
wg.Wait()
}()
}
// TestSharedInformerWatchDisruption simulates a watch that was closed
@@ -371,7 +390,7 @@ func TestSharedInformerInitializationRace(t *testing.T) {
// resync and no resync see the expected state.
func TestSharedInformerWatchDisruption(t *testing.T) {
// source simulates an apiserver object endpoint.
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1", UID: "pod1", ResourceVersion: "1"}})
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod2", UID: "pod2", ResourceVersion: "2"}})
@@ -391,10 +410,13 @@ func TestSharedInformerWatchDisruption(t *testing.T) {
informer.AddEventHandlerWithResyncPeriod(listenerResync, listenerResync.resyncPeriod)
listeners := []*testListener{listenerNoResync, listenerResync}
var wg wait.Group
stop := make(chan struct{})
defer close(stop)
go informer.Run(stop)
wg.StartWithChannel(stop, informer.Run)
defer func() {
close(stop)
wg.Wait()
}()
for _, listener := range listeners {
if !listener.ok() {
@@ -446,7 +468,7 @@ func TestSharedInformerWatchDisruption(t *testing.T) {
}
func TestSharedInformerErrorHandling(t *testing.T) {
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
source.ListError = fmt.Errorf("Access Denied")
@@ -457,8 +479,13 @@ func TestSharedInformerErrorHandling(t *testing.T) {
errCh <- err
})
var wg wait.Group
stop := make(chan struct{})
go informer.Run(stop)
wg.StartWithChannel(stop, informer.Run)
defer func() {
close(stop)
wg.Wait()
}()
select {
case err := <-errCh:
@@ -468,13 +495,12 @@ func TestSharedInformerErrorHandling(t *testing.T) {
case <-time.After(time.Second):
t.Errorf("Timeout waiting for error handler call")
}
close(stop)
}
// TestSharedInformerStartRace is a regression test to ensure there is no race between
// Run and SetWatchErrorHandler, and Run and SetTransform.
func TestSharedInformerStartRace(t *testing.T) {
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second).(*sharedIndexInformer)
stop := make(chan struct{})
go func() {
@@ -493,14 +519,17 @@ func TestSharedInformerStartRace(t *testing.T) {
}
}()
go informer.Run(stop)
var wg wait.Group
wg.StartWithChannel(stop, informer.Run)
defer func() {
close(stop)
wg.Wait()
}()
}
func TestSharedInformerTransformer(t *testing.T) {
// source simulates an apiserver object endpoint.
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1", UID: "pod1", ResourceVersion: "1"}})
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod2", UID: "pod2", ResourceVersion: "2"}})
@@ -521,9 +550,13 @@ func TestSharedInformerTransformer(t *testing.T) {
listenerTransformer := newTestListener("listenerTransformer", 0, "POD1", "POD2")
informer.AddEventHandler(listenerTransformer)
var wg wait.Group
stop := make(chan struct{})
go informer.Run(stop)
defer close(stop)
wg.StartWithChannel(stop, informer.Run)
defer func() {
close(stop)
wg.Wait()
}()
if !listenerTransformer.ok() {
t.Errorf("%s: expected %v, got %v", listenerTransformer.name, listenerTransformer.expectedItemNames, listenerTransformer.receivedItemNames)
@@ -531,7 +564,7 @@ func TestSharedInformerTransformer(t *testing.T) {
}
func TestSharedInformerRemoveHandler(t *testing.T) {
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second)
@@ -569,12 +602,12 @@ func TestSharedInformerRemoveHandler(t *testing.T) {
}
func TestSharedInformerRemoveForeignHandler(t *testing.T) {
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second).(*sharedIndexInformer)
source2 := fcache.NewFakeControllerSource()
source2 := newFakeControllerSource(t)
source2.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
informer2 := NewSharedInformer(source2, &v1.Pod{}, 1*time.Second).(*sharedIndexInformer)
@@ -651,7 +684,7 @@ func TestSharedInformerRemoveForeignHandler(t *testing.T) {
}
func TestSharedInformerMultipleRegistration(t *testing.T) {
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second).(*sharedIndexInformer)
@@ -719,7 +752,7 @@ func TestSharedInformerMultipleRegistration(t *testing.T) {
}
func TestRemovingRemovedSharedInformer(t *testing.T) {
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second).(*sharedIndexInformer)
@@ -751,14 +784,16 @@ func TestRemovingRemovedSharedInformer(t *testing.T) {
// listeners without tripping it up. There are not really many assertions in this
// test. Meant to be run with -race to find race conditions
func TestSharedInformerHandlerAbuse(t *testing.T) {
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second).(*sharedIndexInformer)
ctx, cancel := context.WithCancel(context.Background())
informerCtx, informerCancel := context.WithCancel(context.Background())
go func() {
informer.Run(informerCtx.Done())
var informerWg wait.Group
informerWg.StartWithChannel(informerCtx.Done(), informer.Run)
defer func() {
cancel()
informerWg.Wait()
}()
worker := func() {
@@ -865,7 +900,7 @@ func TestSharedInformerHandlerAbuse(t *testing.T) {
}
func TestStateSharedInformer(t *testing.T) {
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second).(*sharedIndexInformer)
@@ -880,8 +915,10 @@ func TestStateSharedInformer(t *testing.T) {
t.Errorf("informer already stopped after creation")
return
}
var wg wait.Group
stop := make(chan struct{})
go informer.Run(stop)
wg.StartWithChannel(stop, informer.Run)
defer wg.Wait()
if !listener.ok() {
t.Errorf("informer did not report initial objects")
close(stop)
@@ -914,13 +951,15 @@ func TestStateSharedInformer(t *testing.T) {
}
func TestAddOnStoppedSharedInformer(t *testing.T) {
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second).(*sharedIndexInformer)
listener := newTestListener("listener", 0, "pod1")
stop := make(chan struct{})
go informer.Run(stop)
var wg wait.Group
wg.StartWithChannel(stop, informer.Run)
defer wg.Wait()
close(stop)
err := wait.PollImmediate(100*time.Millisecond, 2*time.Second, func() (bool, error) {
@@ -947,7 +986,7 @@ func TestAddOnStoppedSharedInformer(t *testing.T) {
}
func TestRemoveOnStoppedSharedInformer(t *testing.T) {
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second).(*sharedIndexInformer)
@@ -958,7 +997,9 @@ func TestRemoveOnStoppedSharedInformer(t *testing.T) {
return
}
stop := make(chan struct{})
go informer.Run(stop)
var wg wait.Group
wg.StartWithChannel(stop, informer.Run)
defer wg.Wait()
close(stop)
fmt.Println("sleeping")
time.Sleep(1 * time.Second)
@@ -976,7 +1017,7 @@ func TestRemoveOnStoppedSharedInformer(t *testing.T) {
func TestRemoveWhileActive(t *testing.T) {
// source simulates an apiserver object endpoint.
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
// create the shared informer and resync every 12 hours
informer := NewSharedInformer(source, &v1.Pod{}, 0).(*sharedIndexInformer)
@@ -985,9 +1026,13 @@ func TestRemoveWhileActive(t *testing.T) {
handle, _ := informer.AddEventHandler(listener)
stop := make(chan struct{})
defer close(stop)
var wg wait.Group
wg.StartWithChannel(stop, informer.Run)
defer func() {
close(stop)
wg.Wait()
}()
go informer.Run(stop)
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
if !listener.ok() {
@@ -1012,7 +1057,7 @@ func TestRemoveWhileActive(t *testing.T) {
func TestAddWhileActive(t *testing.T) {
// source simulates an apiserver object endpoint.
source := fcache.NewFakeControllerSource()
source := newFakeControllerSource(t)
// create the shared informer and resync every 12 hours
informer := NewSharedInformer(source, &v1.Pod{}, 0).(*sharedIndexInformer)
@@ -1025,7 +1070,12 @@ func TestAddWhileActive(t *testing.T) {
}
stop := make(chan struct{})
defer close(stop)
var wg wait.Group
wg.StartWithChannel(stop, informer.Run)
defer func() {
close(stop)
wg.Wait()
}()
go informer.Run(stop)
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
@@ -1072,3 +1122,212 @@ func TestAddWhileActive(t *testing.T) {
return
}
}
// TestShutdown depends on goleak.VerifyTestMain in main_test.go to verify that
// all goroutines really have stopped in the different scenarios.
func TestShutdown(t *testing.T) {
t.Run("no-context", func(t *testing.T) {
source := newFakeControllerSource(t)
informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second)
handler, err := informer.AddEventHandler(ResourceEventHandlerFuncs{
AddFunc: func(_ any) {},
})
require.NoError(t, err)
defer func() {
assert.NoError(t, informer.RemoveEventHandler(handler))
}()
var wg wait.Group
stop := make(chan struct{})
wg.StartWithChannel(stop, informer.Run)
defer func() {
close(stop)
wg.Wait()
}()
require.Eventually(t, informer.HasSynced, time.Minute, time.Millisecond, "informer has synced")
})
t.Run("no-context-later", func(t *testing.T) {
source := newFakeControllerSource(t)
informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second)
var wg wait.Group
stop := make(chan struct{})
wg.StartWithChannel(stop, informer.Run)
defer func() {
close(stop)
wg.Wait()
}()
require.Eventually(t, informer.HasSynced, time.Minute, time.Millisecond, "informer has synced")
handler, err := informer.AddEventHandler(ResourceEventHandlerFuncs{
AddFunc: func(_ any) {},
})
require.NoError(t, err)
assert.NoError(t, informer.RemoveEventHandler(handler))
})
t.Run("no-run", func(t *testing.T) {
source := newFakeControllerSource(t)
informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second)
_, err := informer.AddEventHandler(ResourceEventHandlerFuncs{
AddFunc: func(_ any) {},
})
require.NoError(t, err)
// At this point, neither informer nor handler have any goroutines running
// and it doesn't matter that nothing gets stopped or removed.
})
}
func TestEventPanics(t *testing.T) {
// timeInUTC := time.Date(2009, 12, 1, 13, 30, 40, 42000, time.UTC)
// timeString := "1201 13:30:40.000042"
// Initialized by init.
var (
buffer threadSafeBuffer
logger klog.Logger
source *fcache.FakeControllerSource
)
init := func(t *testing.T) {
// Restoring state is very sensitive to ordering. All goroutines spawned
// by a test must have completed and there has to be a check that they
// have completed that is visible to the race detector. This also
// applies to all other tests!
t.Cleanup(klog.CaptureState().Restore) //nolint:logcheck // CaptureState shouldn't be used in packages with contextual logging, but here it is okay.
buffer.buffer.Reset()
logger = textlogger.NewLogger(textlogger.NewConfig(
// textlogger.FixedTime(timeInUTC),
textlogger.Output(&buffer),
))
oldReallyCrash := utilruntime.ReallyCrash
utilruntime.ReallyCrash = false
t.Cleanup(func() { utilruntime.ReallyCrash = oldReallyCrash })
source = newFakeControllerSource(t)
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
}
newHandler := func(ctx context.Context) ResourceEventHandlerFuncs {
logger := klog.FromContext(ctx)
return ResourceEventHandlerFuncs{
AddFunc: func(obj any) {
logger.Info("Add func will panic now", "pod", klog.KObj(obj.(*v1.Pod)))
panic("fake panic")
},
}
}
_, _, panicLine, _ := runtime.Caller(0)
panicLine -= 4
expectedLog := func(name string) string {
if name == "" {
return fmt.Sprintf(`shared_informer_test.go:%d] "Observed a panic" panic="fake panic"`, panicLine)
}
return fmt.Sprintf(`shared_informer_test.go:%d] "Observed a panic" logger=%q panic="fake panic"`, panicLine, name)
}
handler := newHandler(context.Background())
t.Run("simple", func(t *testing.T) {
init(t)
klog.SetLogger(logger)
stop := make(chan struct{})
informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second)
handle, err := informer.AddEventHandler(handler)
require.NoError(t, err)
defer func() {
assert.NoError(t, informer.RemoveEventHandler(handle))
}()
var wg wait.Group
wg.StartWithChannel(stop, informer.Run)
defer func() {
close(stop)
assert.Eventually(t, informer.IsStopped, time.Minute, time.Millisecond, "informer has stopped")
wg.Wait() // For race detector...
}()
require.Eventually(t, informer.HasSynced, time.Minute, time.Millisecond, "informer has synced")
// This times out (https://github.com/kubernetes/kubernetes/issues/129024) because the
// handler never syncs when the callback panics:
// require.Eventually(t, handle.HasSynced, time.Minute, time.Millisecond, "handler has synced")
//
// Wait for a non-empty buffer instead. This implies that we have to make
// the buffer thread-safe, which wouldn't be necessary otherwise.
assert.EventuallyWithT(t, func(t *assert.CollectT) {
assert.Contains(t, buffer.String(), expectedLog(""))
}, time.Minute, time.Millisecond, "handler has panicked")
})
t.Run("many", func(t *testing.T) {
init(t)
// One pod was already created in init, add some more.
numPods := 5
for i := 1; i < numPods; i++ {
source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("pod%d", i+1)}})
}
_, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancel(ctx)
defer cancel()
informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second)
name1 := "fake-event-handler-1"
logger1 := klog.LoggerWithName(logger, name1)
ctx1 := klog.NewContext(ctx, logger1)
handle1, err := informer.AddEventHandlerWithOptions(newHandler(ctx1), HandlerOptions{Logger: &logger1})
require.NoError(t, err)
defer func() {
assert.NoError(t, informer.RemoveEventHandler(handle1))
}()
name2 := "fake-event-handler-2"
logger2 := klog.LoggerWithName(logger, name2)
ctx2 := klog.NewContext(ctx, logger2)
handle2, err := informer.AddEventHandlerWithOptions(newHandler(ctx2), HandlerOptions{Logger: &logger2})
require.NoError(t, err)
defer func() {
assert.NoError(t, informer.RemoveEventHandler(handle2))
}()
start := time.Now()
var wg wait.Group
informerName := "informer"
informerLogger := klog.LoggerWithName(logger, informerName)
informerCtx := klog.NewContext(ctx, informerLogger)
wg.StartWithContext(informerCtx, informer.RunWithContext)
defer func() {
cancel()
assert.Eventually(t, informer.IsStopped, time.Minute, time.Millisecond, "informer has stopped")
wg.Wait() // For race detector...
}()
require.Eventually(t, informer.HasSynced, time.Minute, time.Millisecond, "informer has synced")
assert.EventuallyWithT(t, func(t *assert.CollectT) {
output := buffer.String()
expected := expectedLog(name1)
if !assert.Equal(t, numPods, numOccurrences(output, expected), "Log output should have the right number of panics for %q (search string: %q), got instead:\n%s", name1, expected, output) {
return
}
expected = expectedLog(name2)
assert.Equal(t, numPods, numOccurrences(output, expected), "Log output should have the right number of panics for %q (search string %q, got instead:\n%s", name2, expected, output)
}, 30*time.Second, time.Millisecond, "handler has panicked")
// Both handlers should have slept for one second after each panic,
// except after the last pod event because then the input channel
// gets closed.
assert.GreaterOrEqual(t, time.Since(start), time.Duration(numPods-1)*time.Second, "Delay in processorListener.run")
})
}
func numOccurrences(hay, needle string) int {
count := 0
for {
index := strings.Index(hay, needle)
if index < 0 {
return count
}
count++
hay = hay[index+len(needle):]
}
}

49
staging/src/k8s.io/client-go/tools/cache/util_test.go vendored Normal file
View File

@@ -0,0 +1,49 @@
/*
Copyright 2024 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package cache
import (
"bytes"
"sync"
"testing"
fcache "k8s.io/client-go/tools/cache/testing"
)
func newFakeControllerSource(tb testing.TB) *fcache.FakeControllerSource {
source := fcache.NewFakeControllerSource()
tb.Cleanup(source.Shutdown)
return source
}
// threadSafeBuffer is a thread-safe wrapper around bytes.Buffer.
type threadSafeBuffer struct {
buffer bytes.Buffer
mu sync.Mutex
}
func (b *threadSafeBuffer) Write(p []byte) (n int, err error) {
b.mu.Lock()
defer b.mu.Unlock()
return b.buffer.Write(p)
}
func (b *threadSafeBuffer) String() string {
b.mu.Lock()
defer b.mu.Unlock()
return b.buffer.String()
}

2
staging/src/k8s.io/code-generator/examples/go.sum generated
View File

@@ -81,6 +81,8 @@ github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=

1
staging/src/k8s.io/component-helpers/go.sum generated
View File

@@ -92,6 +92,7 @@ github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=

2
staging/src/k8s.io/dynamic-resource-allocation/resourceslice/resourceslicecontroller.go
View File

@@ -325,7 +325,7 @@ func (c *Controller) initInformer(ctx context.Context) error {
}, func(options *metav1.ListOptions) {
options.FieldSelector = selector.String()
})
c.sliceStore = cache.NewIntegerResourceVersionMutationCache(informer.GetStore(), informer.GetIndexer(), c.mutationCacheTTL, true /* includeAdds */)
c.sliceStore = cache.NewIntegerResourceVersionMutationCache(logger, informer.GetStore(), informer.GetIndexer(), c.mutationCacheTTL, true /* includeAdds */)
handler, err := informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: func(obj any) {
slice, ok := obj.(*resourceapi.ResourceSlice)

2
test/integration/serviceaccount/service_account_test.go
View File

@@ -38,6 +38,7 @@ import (
clientset "k8s.io/client-go/kubernetes"
restclient "k8s.io/client-go/rest"
"k8s.io/client-go/util/keyutil"
"k8s.io/klog/v2"
"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
"k8s.io/kubernetes/pkg/controller"
serviceaccountcontroller "k8s.io/kubernetes/pkg/controller/serviceaccount"
@@ -393,6 +394,7 @@ func startServiceAccountTestServerAndWaitForCaches(ctx context.Context, t *testi
return rootClientset, clientConfig, stop, informers, err
}
tokenController, err := serviceaccountcontroller.NewTokensController(
klog.FromContext(ctx),
informers.Core().V1().ServiceAccounts(),
informers.Core().V1().Secrets(),
rootClientset,