mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-07-23 03:41:45 +00:00
Merge pull request #18080 from wojtek-t/list_options_in_listwatch
Pass ListOptions to List in ListWatch.
This commit is contained in:
commit
a915b8b29a
@ -523,7 +523,7 @@ func NewMockPodsListWatch(initialPodList api.PodList) *MockPodsListWatch {
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return lw.fakeWatcher, nil
|
||||
},
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return &lw.list, nil
|
||||
},
|
||||
}
|
||||
|
@ -173,7 +173,7 @@ func NewMockPodsListWatch(initialPodList api.PodList) *MockPodsListWatch {
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return lw.fakeWatcher, nil
|
||||
},
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
lw.lock.Lock()
|
||||
defer lw.lock.Unlock()
|
||||
|
||||
|
@ -58,8 +58,8 @@ func NewEndpointController(client *client.Client) *endpointController {
|
||||
}
|
||||
e.serviceStore.Store, e.serviceController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return e.client.Services(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return e.client.Services(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return e.client.Services(api.NamespaceAll).Watch(options)
|
||||
@ -78,8 +78,8 @@ func NewEndpointController(client *client.Client) *endpointController {
|
||||
|
||||
e.podStore.Store, e.podController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return e.client.Pods(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return e.client.Pods(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return e.client.Pods(api.NamespaceAll).Watch(options)
|
||||
|
9
pkg/client/cache/listwatch.go
vendored
9
pkg/client/cache/listwatch.go
vendored
@ -28,7 +28,7 @@ import (
|
||||
)
|
||||
|
||||
// ListFunc knows how to list resources
|
||||
type ListFunc func() (runtime.Object, error)
|
||||
type ListFunc func(options unversioned.ListOptions) (runtime.Object, error)
|
||||
|
||||
// WatchFunc knows how to watch resources
|
||||
type WatchFunc func(options unversioned.ListOptions) (watch.Interface, error)
|
||||
@ -48,10 +48,11 @@ type Getter interface {
|
||||
|
||||
// NewListWatchFromClient creates a new ListWatch from the specified client, resource, namespace and field selector.
|
||||
func NewListWatchFromClient(c Getter, resource string, namespace string, fieldSelector fields.Selector) *ListWatch {
|
||||
listFunc := func() (runtime.Object, error) {
|
||||
listFunc := func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return c.Get().
|
||||
Namespace(namespace).
|
||||
Resource(resource).
|
||||
VersionedParams(&options, api.Scheme).
|
||||
FieldsSelectorParam(fieldSelector).
|
||||
Do().
|
||||
Get()
|
||||
@ -76,8 +77,8 @@ func timeoutFromListOptions(options unversioned.ListOptions) time.Duration {
|
||||
}
|
||||
|
||||
// List a set of apiserver resources
|
||||
func (lw *ListWatch) List() (runtime.Object, error) {
|
||||
return lw.ListFunc()
|
||||
func (lw *ListWatch) List(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return lw.ListFunc(options)
|
||||
}
|
||||
|
||||
// Watch a set of apiserver resources
|
||||
|
2
pkg/client/cache/listwatch_test.go
vendored
2
pkg/client/cache/listwatch_test.go
vendored
@ -99,7 +99,7 @@ func TestListWatchesCanList(t *testing.T) {
|
||||
client := client.NewOrDie(&client.Config{Host: server.URL, GroupVersion: testapi.Default.GroupVersion()})
|
||||
lw := NewListWatchFromClient(client, item.resource, item.namespace, item.fieldSelector)
|
||||
// This test merely tests that the correct request is made.
|
||||
lw.List()
|
||||
lw.List(unversioned.ListOptions{})
|
||||
handler.ValidateRequest(t, item.location, "GET", nil)
|
||||
}
|
||||
}
|
||||
|
5
pkg/client/cache/reflector.go
vendored
5
pkg/client/cache/reflector.go
vendored
@ -43,7 +43,7 @@ import (
|
||||
type ListerWatcher interface {
|
||||
// List should return a list type object; the Items field will be extracted, and the
|
||||
// ResourceVersion field will be used to start the watch in the right place.
|
||||
List() (runtime.Object, error)
|
||||
List(options unversioned.ListOptions) (runtime.Object, error)
|
||||
// Watch should begin a watch at the specified version.
|
||||
Watch(options unversioned.ListOptions) (watch.Interface, error)
|
||||
}
|
||||
@ -227,7 +227,8 @@ func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
|
||||
resyncCh, cleanup := r.resyncChan()
|
||||
defer cleanup()
|
||||
|
||||
list, err := r.listerWatcher.List()
|
||||
options := unversioned.ListOptions{}
|
||||
list, err := r.listerWatcher.List(options)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s: Failed to list %v: %v", r.name, r.expectedType, err)
|
||||
}
|
||||
|
4
pkg/client/cache/reflector_test.go
vendored
4
pkg/client/cache/reflector_test.go
vendored
@ -35,7 +35,9 @@ type testLW struct {
|
||||
WatchFunc func(resourceVersion string) (watch.Interface, error)
|
||||
}
|
||||
|
||||
func (t *testLW) List() (runtime.Object, error) { return t.ListFunc() }
|
||||
func (t *testLW) List(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return t.ListFunc()
|
||||
}
|
||||
func (t *testLW) Watch(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return t.WatchFunc(options.ResourceVersion)
|
||||
}
|
||||
|
@ -96,8 +96,8 @@ func NewDaemonSetsController(kubeClient client.Interface, resyncPeriod controlle
|
||||
// Manage addition/update of daemon sets.
|
||||
dsc.dsStore.Store, dsc.dsController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return dsc.kubeClient.Extensions().DaemonSets(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return dsc.kubeClient.Extensions().DaemonSets(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return dsc.kubeClient.Extensions().DaemonSets(api.NamespaceAll).Watch(options)
|
||||
@ -128,8 +128,8 @@ func NewDaemonSetsController(kubeClient client.Interface, resyncPeriod controlle
|
||||
// more pods until all the effects (expectations) of a daemon set's create/delete have been observed.
|
||||
dsc.podStore.Store, dsc.podController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return dsc.kubeClient.Pods(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return dsc.kubeClient.Pods(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return dsc.kubeClient.Pods(api.NamespaceAll).Watch(options)
|
||||
@ -146,8 +146,8 @@ func NewDaemonSetsController(kubeClient client.Interface, resyncPeriod controlle
|
||||
// Watch for new nodes or updates to nodes - daemon pods are launched on new nodes, and possibly when labels on nodes change,
|
||||
dsc.nodeStore.Store, dsc.nodeController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return dsc.kubeClient.Nodes().List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return dsc.kubeClient.Nodes().List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return dsc.kubeClient.Nodes().Watch(options)
|
||||
|
@ -62,8 +62,8 @@ func NewEndpointController(client *client.Client, resyncPeriod controller.Resync
|
||||
|
||||
e.serviceStore.Store, e.serviceController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return e.client.Services(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return e.client.Services(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return e.client.Services(api.NamespaceAll).Watch(options)
|
||||
@ -83,8 +83,8 @@ func NewEndpointController(client *client.Client, resyncPeriod controller.Resync
|
||||
|
||||
e.podStore.Store, e.podController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return e.client.Pods(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return e.client.Pods(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return e.client.Pods(api.NamespaceAll).Watch(options)
|
||||
|
@ -122,7 +122,7 @@ func (f *FakeControllerSource) Change(e watch.Event, watchProbability float64) {
|
||||
}
|
||||
|
||||
// List returns a list object, with its resource version set.
|
||||
func (f *FakeControllerSource) List() (runtime.Object, error) {
|
||||
func (f *FakeControllerSource) List(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
f.lock.RLock()
|
||||
defer f.lock.RUnlock()
|
||||
list := make([]runtime.Object, 0, len(f.items))
|
||||
|
@ -71,7 +71,7 @@ func TestRCNumber(t *testing.T) {
|
||||
}
|
||||
go consume(t, w, []string{"2", "3"}, wg)
|
||||
|
||||
list, err := source.List()
|
||||
list, err := source.List(unversioned.ListOptions{})
|
||||
if err != nil {
|
||||
t.Fatalf("Unexpected error: %v", err)
|
||||
}
|
||||
|
@ -66,8 +66,8 @@ func New(kubeClient client.Interface, resyncPeriod controller.ResyncPeriodFunc,
|
||||
|
||||
gcc.podStore.Store, gcc.podStoreSyncer = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
options := unversioned.ListOptions{FieldSelector: unversioned.FieldSelector{terminatedSelector}}
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
options.FieldSelector.Selector = terminatedSelector
|
||||
return gcc.kubeClient.Pods(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
|
@ -83,8 +83,8 @@ func NewJobController(kubeClient client.Interface, resyncPeriod controller.Resyn
|
||||
|
||||
jm.jobStore.Store, jm.jobController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return jm.kubeClient.Extensions().Jobs(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return jm.kubeClient.Extensions().Jobs(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return jm.kubeClient.Extensions().Jobs(api.NamespaceAll).Watch(options)
|
||||
@ -106,8 +106,8 @@ func NewJobController(kubeClient client.Interface, resyncPeriod controller.Resyn
|
||||
|
||||
jm.podStore.Store, jm.podController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return jm.kubeClient.Pods(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return jm.kubeClient.Pods(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return jm.kubeClient.Pods(api.NamespaceAll).Watch(options)
|
||||
|
@ -45,8 +45,8 @@ func NewNamespaceController(kubeClient client.Interface, versions *unversioned.A
|
||||
var controller *framework.Controller
|
||||
_, controller = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return kubeClient.Namespaces().List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return kubeClient.Namespaces().List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return kubeClient.Namespaces().Watch(options)
|
||||
|
@ -162,8 +162,8 @@ func NewNodeController(
|
||||
|
||||
nc.podStore.Store, nc.podController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return nc.kubeClient.Pods(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return nc.kubeClient.Pods(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return nc.kubeClient.Pods(api.NamespaceAll).Watch(options)
|
||||
@ -178,8 +178,8 @@ func NewNodeController(
|
||||
)
|
||||
nc.nodeStore.Store, nc.nodeController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return nc.kubeClient.Nodes().List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return nc.kubeClient.Nodes().List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return nc.kubeClient.Nodes().Watch(options)
|
||||
|
@ -55,8 +55,8 @@ func NewPersistentVolumeClaimBinder(kubeClient client.Interface, syncPeriod time
|
||||
|
||||
_, volumeController := framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return kubeClient.PersistentVolumes().List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return kubeClient.PersistentVolumes().List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return kubeClient.PersistentVolumes().Watch(options)
|
||||
@ -73,8 +73,8 @@ func NewPersistentVolumeClaimBinder(kubeClient client.Interface, syncPeriod time
|
||||
)
|
||||
_, claimController := framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return kubeClient.PersistentVolumeClaims(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return kubeClient.PersistentVolumeClaims(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return kubeClient.PersistentVolumeClaims(api.NamespaceAll).Watch(options)
|
||||
|
@ -62,8 +62,8 @@ func NewPersistentVolumeRecycler(kubeClient client.Interface, syncPeriod time.Du
|
||||
|
||||
_, volumeController := framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return kubeClient.PersistentVolumes().List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return kubeClient.PersistentVolumes().List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return kubeClient.PersistentVolumes().Watch(options)
|
||||
|
@ -107,8 +107,8 @@ func NewReplicationManager(kubeClient client.Interface, resyncPeriod controller.
|
||||
|
||||
rm.rcStore.Store, rm.rcController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return rm.kubeClient.ReplicationControllers(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return rm.kubeClient.ReplicationControllers(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return rm.kubeClient.ReplicationControllers(api.NamespaceAll).Watch(options)
|
||||
@ -148,8 +148,8 @@ func NewReplicationManager(kubeClient client.Interface, resyncPeriod controller.
|
||||
|
||||
rm.podStore.Store, rm.podController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return rm.kubeClient.Pods(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return rm.kubeClient.Pods(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return rm.kubeClient.Pods(api.NamespaceAll).Watch(options)
|
||||
|
@ -65,8 +65,8 @@ func NewResourceQuotaController(kubeClient client.Interface, resyncPeriod contro
|
||||
|
||||
rq.rqIndexer, rq.rqController = framework.NewIndexerInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return rq.kubeClient.ResourceQuotas(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return rq.kubeClient.ResourceQuotas(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return rq.kubeClient.ResourceQuotas(api.NamespaceAll).Watch(options)
|
||||
@ -105,8 +105,8 @@ func NewResourceQuotaController(kubeClient client.Interface, resyncPeriod contro
|
||||
// release compute resources from any associated quota.
|
||||
rq.podStore.Store, rq.podController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return rq.kubeClient.Pods(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return rq.kubeClient.Pods(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return rq.kubeClient.Pods(api.NamespaceAll).Watch(options)
|
||||
|
@ -80,8 +80,8 @@ func NewServiceAccountsController(cl client.Interface, options ServiceAccountsCo
|
||||
}
|
||||
e.serviceAccounts, e.serviceAccountController = framework.NewIndexerInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
options := unversioned.ListOptions{FieldSelector: unversioned.FieldSelector{accountSelector}}
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
options.FieldSelector.Selector = accountSelector
|
||||
return e.client.ServiceAccounts(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
@ -99,8 +99,8 @@ func NewServiceAccountsController(cl client.Interface, options ServiceAccountsCo
|
||||
|
||||
e.namespaces, e.namespaceController = framework.NewIndexerInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return e.client.Namespaces().List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return e.client.Namespaces().List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return e.client.Namespaces().Watch(options)
|
||||
|
@ -62,8 +62,8 @@ func NewTokensController(cl client.Interface, options TokensControllerOptions) *
|
||||
|
||||
e.serviceAccounts, e.serviceAccountController = framework.NewIndexerInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return e.client.ServiceAccounts(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return e.client.ServiceAccounts(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return e.client.ServiceAccounts(api.NamespaceAll).Watch(options)
|
||||
@ -82,8 +82,8 @@ func NewTokensController(cl client.Interface, options TokensControllerOptions) *
|
||||
tokenSelector := fields.SelectorFromSet(map[string]string{client.SecretType: string(api.SecretTypeServiceAccountToken)})
|
||||
e.secrets, e.secretController = framework.NewIndexerInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
options := unversioned.ListOptions{FieldSelector: unversioned.FieldSelector{tokenSelector}}
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
options.FieldSelector.Selector = tokenSelector
|
||||
return e.client.Secrets(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
|
@ -32,7 +32,7 @@ type fakePodLW struct {
|
||||
watchResp watch.Interface
|
||||
}
|
||||
|
||||
func (lw fakePodLW) List() (runtime.Object, error) {
|
||||
func (lw fakePodLW) List(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return lw.listResp, nil
|
||||
}
|
||||
|
||||
|
@ -233,8 +233,8 @@ func NewMainKubelet(
|
||||
// TODO: cache.NewListWatchFromClient is limited as it takes a client implementation rather
|
||||
// than an interface. There is no way to construct a list+watcher using resource name.
|
||||
listWatch := &cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return kubeClient.Services(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return kubeClient.Services(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return kubeClient.Services(api.NamespaceAll).Watch(options)
|
||||
@ -250,8 +250,8 @@ func NewMainKubelet(
|
||||
// than an interface. There is no way to construct a list+watcher using resource name.
|
||||
fieldSelector := fields.Set{client.ObjectNameField: nodeName}.AsSelector()
|
||||
listWatch := &cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
options := unversioned.ListOptions{FieldSelector: unversioned.FieldSelector{fieldSelector}}
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
options.FieldSelector.Selector = fieldSelector
|
||||
return kubeClient.Nodes().List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
|
@ -32,7 +32,7 @@ type fakeLW struct {
|
||||
watchResp watch.Interface
|
||||
}
|
||||
|
||||
func (lw fakeLW) List() (runtime.Object, error) {
|
||||
func (lw fakeLW) List(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return lw.listResp, nil
|
||||
}
|
||||
|
||||
|
@ -387,7 +387,7 @@ func newCacherListerWatcher(storage Interface, resourcePrefix string, newListFun
|
||||
}
|
||||
|
||||
// Implements cache.ListerWatcher interface.
|
||||
func (lw *cacherListerWatcher) List() (runtime.Object, error) {
|
||||
func (lw *cacherListerWatcher) List(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
list := lw.newListFunc()
|
||||
if err := lw.storage.List(context.TODO(), lw.resourcePrefix, 0, Everything, list); err != nil {
|
||||
return nil, err
|
||||
|
@ -249,11 +249,13 @@ func TestWaitUntilFreshAndList(t *testing.T) {
|
||||
}
|
||||
|
||||
type testLW struct {
|
||||
ListFunc func() (runtime.Object, error)
|
||||
ListFunc func(options unversioned.ListOptions) (runtime.Object, error)
|
||||
WatchFunc func(options unversioned.ListOptions) (watch.Interface, error)
|
||||
}
|
||||
|
||||
func (t *testLW) List() (runtime.Object, error) { return t.ListFunc() }
|
||||
func (t *testLW) List(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return t.ListFunc(options)
|
||||
}
|
||||
func (t *testLW) Watch(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return t.WatchFunc(options)
|
||||
}
|
||||
@ -274,7 +276,7 @@ func TestReflectorForWatchCache(t *testing.T) {
|
||||
go fw.Stop()
|
||||
return fw, nil
|
||||
},
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return &api.PodList{ListMeta: unversioned.ListMeta{ResourceVersion: "10"}}, nil
|
||||
},
|
||||
}
|
||||
|
@ -108,8 +108,8 @@ func (c *realRecyclerClient) WatchPod(name, namespace, resourceVersion string, s
|
||||
fieldSelector, _ := fields.ParseSelector("metadata.name=" + name)
|
||||
|
||||
podLW := &cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
options := unversioned.ListOptions{FieldSelector: unversioned.FieldSelector{fieldSelector}}
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
options.FieldSelector.Selector = fieldSelector
|
||||
return c.client.Pods(namespace).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
|
@ -97,8 +97,8 @@ func (l *limitRanger) Admit(a admission.Attributes) (err error) {
|
||||
// NewLimitRanger returns an object that enforces limits based on the supplied limit function
|
||||
func NewLimitRanger(client client.Interface, limitFunc LimitFunc) admission.Interface {
|
||||
lw := &cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return client.LimitRanges(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return client.LimitRanges(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return client.LimitRanges(api.NamespaceAll).Watch(options)
|
||||
|
@ -83,8 +83,8 @@ func NewProvision(c client.Interface) admission.Interface {
|
||||
store := cache.NewStore(cache.MetaNamespaceKeyFunc)
|
||||
reflector := cache.NewReflector(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return c.Namespaces().List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return c.Namespaces().List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return c.Namespaces().Watch(options)
|
||||
|
@ -90,8 +90,8 @@ func NewExists(c client.Interface) admission.Interface {
|
||||
store := cache.NewStore(cache.MetaNamespaceKeyFunc)
|
||||
reflector := cache.NewReflector(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return c.Namespaces().List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return c.Namespaces().List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return c.Namespaces().Watch(options)
|
||||
|
@ -107,8 +107,8 @@ func NewLifecycle(c client.Interface) admission.Interface {
|
||||
store := cache.NewStore(cache.MetaNamespaceKeyFunc)
|
||||
reflector := cache.NewReflector(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return c.Namespaces().List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return c.Namespaces().List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return c.Namespaces().Watch(options)
|
||||
|
@ -49,8 +49,8 @@ type quota struct {
|
||||
// NewResourceQuota creates a new resource quota admission control handler
|
||||
func NewResourceQuota(client client.Interface) admission.Interface {
|
||||
lw := &cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return client.ResourceQuotas(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return client.ResourceQuotas(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return client.ResourceQuotas(api.NamespaceAll).Watch(options)
|
||||
|
@ -90,8 +90,8 @@ type serviceAccount struct {
|
||||
func NewServiceAccount(cl client.Interface) *serviceAccount {
|
||||
serviceAccountsIndexer, serviceAccountsReflector := cache.NewNamespaceKeyedIndexerAndReflector(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return cl.ServiceAccounts(api.NamespaceAll).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return cl.ServiceAccounts(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return cl.ServiceAccounts(api.NamespaceAll).Watch(options)
|
||||
@ -104,8 +104,8 @@ func NewServiceAccount(cl client.Interface) *serviceAccount {
|
||||
tokenSelector := fields.SelectorFromSet(map[string]string{client.SecretType: string(api.SecretTypeServiceAccountToken)})
|
||||
secretsIndexer, secretsReflector := cache.NewNamespaceKeyedIndexerAndReflector(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
options := unversioned.ListOptions{FieldSelector: unversioned.FieldSelector{tokenSelector}}
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
options.FieldSelector.Selector = tokenSelector
|
||||
return cl.Secrets(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
|
@ -220,8 +220,8 @@ var _ = Describe("DaemonRestart", func() {
|
||||
tracker = newPodTracker()
|
||||
newPods, controller = controllerframework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
options := unversioned.ListOptions{LabelSelector: unversioned.LabelSelector{labelSelector}}
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
options.LabelSelector.Selector = labelSelector
|
||||
return framework.Client.Pods(ns).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
|
@ -235,8 +235,8 @@ var _ = Describe("Density [Skipped]", func() {
|
||||
events := make([](*api.Event), 0)
|
||||
_, controller := controllerframework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return c.Events(ns).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return c.Events(ns).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return c.Events(ns).Watch(options)
|
||||
@ -318,9 +318,9 @@ var _ = Describe("Density [Skipped]", func() {
|
||||
additionalPodsPrefix = "density-latency-pod-" + string(util.NewUUID())
|
||||
_, controller := controllerframework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
selector := labels.SelectorFromSet(labels.Set{"name": additionalPodsPrefix})
|
||||
options := unversioned.ListOptions{LabelSelector: unversioned.LabelSelector{selector}}
|
||||
options.LabelSelector.Selector = selector
|
||||
return c.Pods(ns).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
|
@ -143,9 +143,9 @@ func runLatencyTest(nodeCount int, c *client.Client, ns string) {
|
||||
stopCh := make(chan struct{})
|
||||
_, informer := framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
selector := labels.SelectorFromSet(labels.Set{"name": additionalPodsPrefix})
|
||||
options := unversioned.ListOptions{LabelSelector: unversioned.LabelSelector{selector}}
|
||||
options.LabelSelector.Selector = selector
|
||||
return c.Pods(ns).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
|
@ -277,8 +277,8 @@ func (eq *endpointQueries) added(e *api.Endpoints) {
|
||||
func startEndpointWatcher(f *Framework, q *endpointQueries) {
|
||||
_, controller := framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
return f.Client.Endpoints(f.Namespace.Name).List(unversioned.ListOptions{})
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
return f.Client.Endpoints(f.Namespace.Name).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
return f.Client.Endpoints(f.Namespace.Name).Watch(options)
|
||||
|
@ -161,11 +161,9 @@ type podStore struct {
|
||||
|
||||
func newPodStore(c *client.Client, namespace string, label labels.Selector, field fields.Selector) *podStore {
|
||||
lw := &cache.ListWatch{
|
||||
ListFunc: func() (runtime.Object, error) {
|
||||
options := unversioned.ListOptions{
|
||||
LabelSelector: unversioned.LabelSelector{label},
|
||||
FieldSelector: unversioned.FieldSelector{field},
|
||||
}
|
||||
ListFunc: func(options unversioned.ListOptions) (runtime.Object, error) {
|
||||
options.LabelSelector.Selector = label
|
||||
options.FieldSelector.Selector = field
|
||||
return c.Pods(namespace).List(options)
|
||||
},
|
||||
WatchFunc: func(options unversioned.ListOptions) (watch.Interface, error) {
|
||||
|
Loading…
Reference in New Issue
Block a user