Skip to content

K
k3s
Commit 0e1ee14c authored by k8s-merge-robot's avatar k8s-merge-robot
Browse files
Options

Merge pull request #15705 from wojtek-t/watch_options_in_client

Auto commit by PR queue bot
parents 33dabe00 aa57f607
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 401 additions and 160 deletions
api/swagger-spec/v1beta1.json
View file @ 0e1ee14c
......@@ -55,6 +55,14 @@
"allowMultiple": false
},
{
"type": "*int64",
"paramType": "query",
"name": "timeoutSeconds",
"description": "Timeout for the list/watch call.",
"required": false,
"allowMultiple": false
},
{
"type": "string",
"paramType": "path",
"name": "namespace",
......@@ -175,6 +183,14 @@
"allowMultiple": false
},
{
"type": "*int64",
"paramType": "query",
"name": "timeoutSeconds",
"description": "Timeout for the list/watch call.",
"required": false,
"allowMultiple": false
},
{
"type": "string",
"paramType": "path",
"name": "namespace",
......@@ -462,6 +478,14 @@
"allowMultiple": false
},
{
"type": "*int64",
"paramType": "query",
"name": "timeoutSeconds",
"description": "Timeout for the list/watch call.",
"required": false,
"allowMultiple": false
},
{
"type": "string",
"paramType": "path",
"name": "namespace",
......@@ -543,6 +567,14 @@
"description": "When specified with a watch call, shows changes that occur after that particular version of a resource. Defaults to changes from the beginning of history.",
"required": false,
"allowMultiple": false
},
{
"type": "*int64",
"paramType": "query",
"name": "timeoutSeconds",
"description": "Timeout for the list/watch call.",
"required": false,
"allowMultiple": false
}
],
"responseMessages": [
......@@ -610,6 +642,14 @@
"description": "When specified with a watch call, shows changes that occur after that particular version of a resource. Defaults to changes from the beginning of history.",
"required": false,
"allowMultiple": false
},
{
"type": "*int64",
"paramType": "query",
"name": "timeoutSeconds",
"description": "Timeout for the list/watch call.",
"required": false,
"allowMultiple": false
}
],
"responseMessages": [
......@@ -738,6 +778,14 @@
"allowMultiple": false
},
{
"type": "*int64",
"paramType": "query",
"name": "timeoutSeconds",
"description": "Timeout for the list/watch call.",
"required": false,
"allowMultiple": false
},
{
"type": "string",
"paramType": "path",
"name": "namespace",
......@@ -858,6 +906,14 @@
"allowMultiple": false
},
{
"type": "*int64",
"paramType": "query",
"name": "timeoutSeconds",
"description": "Timeout for the list/watch call.",
"required": false,
"allowMultiple": false
},
{
"type": "string",
"paramType": "path",
"name": "namespace",
......@@ -1145,6 +1201,14 @@
"allowMultiple": false
},
{
"type": "*int64",
"paramType": "query",
"name": "timeoutSeconds",
"description": "Timeout for the list/watch call.",
"required": false,
"allowMultiple": false
},
{
"type": "string",
"paramType": "path",
"name": "namespace",
......@@ -1226,6 +1290,14 @@
"description": "When specified with a watch call, shows changes that occur after that particular version of a resource. Defaults to changes from the beginning of history.",
"required": false,
"allowMultiple": false
},
{
"type": "*int64",
"paramType": "query",
"name": "timeoutSeconds",
"description": "Timeout for the list/watch call.",
"required": false,
"allowMultiple": false
}
],
"responseMessages": [
......@@ -1293,6 +1365,14 @@
"description": "When specified with a watch call, shows changes that occur after that particular version of a resource. Defaults to changes from the beginning of history.",
"required": false,
"allowMultiple": false
},
{
"type": "*int64",
"paramType": "query",
"name": "timeoutSeconds",
"description": "Timeout for the list/watch call.",
"required": false,
"allowMultiple": false
}
],
"responseMessages": [
......@@ -1421,6 +1501,14 @@
"allowMultiple": false
},
{
"type": "*int64",
"paramType": "query",
"name": "timeoutSeconds",
"description": "Timeout for the list/watch call.",
"required": false,
"allowMultiple": false
},
{
"type": "string",
"paramType": "path",
"name": "namespace",
......@@ -1541,6 +1629,14 @@
"allowMultiple": false
},
{
"type": "*int64",
"paramType": "query",
"name": "timeoutSeconds",
"description": "Timeout for the list/watch call.",
"required": false,
"allowMultiple": false
},
{
"type": "string",
"paramType": "path",
"name": "namespace",
......@@ -1828,6 +1924,14 @@
"allowMultiple": false
},
{
"type": "*int64",
"paramType": "query",
"name": "timeoutSeconds",
"description": "Timeout for the list/watch call.",
"required": false,
"allowMultiple": false
},
{
"type": "string",
"paramType": "path",
"name": "namespace",
......@@ -1909,6 +2013,14 @@
"description": "When specified with a watch call, shows changes that occur after that particular version of a resource. Defaults to changes from the beginning of history.",
"required": false,
"allowMultiple": false
},
{
"type": "*int64",
"paramType": "query",
"name": "timeoutSeconds",
"description": "Timeout for the list/watch call.",
"required": false,
"allowMultiple": false
}
],
"responseMessages": [
......@@ -1976,6 +2088,14 @@
"description": "When specified with a watch call, shows changes that occur after that particular version of a resource. Defaults to changes from the beginning of history.",
"required": false,
"allowMultiple": false
},
{
"type": "*int64",
"paramType": "query",
"name": "timeoutSeconds",
"description": "Timeout for the list/watch call.",
"required": false,
"allowMultiple": false
}
],
"responseMessages": [
......
contrib/mesos/pkg/service/endpoints_controller.go
View file @ 0e1ee14c
......@@ -61,7 +61,8 @@ func NewEndpointController(client *client.Client) *endpointController {
return e.client.Services(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return e.client.Services(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return e.client.Services(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Service{},
......@@ -81,7 +82,8 @@ func NewEndpointController(client *client.Client) *endpointController {
return e.client.Pods(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return e.client.Pods(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return e.client.Pods(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Pod{},
......
pkg/api/deep_copy_generated.go
View file @ 0e1ee14c
......@@ -826,6 +826,12 @@ func deepCopy_api_ListOptions(in ListOptions, out *ListOptions, c *conversion.Cl
}
out.Watch = in.Watch
out.ResourceVersion = in.ResourceVersion
if in.TimeoutSeconds != nil {
out.TimeoutSeconds = new(int64)
*out.TimeoutSeconds = *in.TimeoutSeconds
} else {
out.TimeoutSeconds = nil
}
return nil
}
......
pkg/api/types.generated.go
View file @ 0e1ee14c
This source diff could not be displayed because it is too large. You can view the blob instead.
pkg/api/types.go
View file @ 0e1ee14c
......@@ -1645,6 +1645,8 @@ type ListOptions struct {
Watch bool
// The resource version to watch (no effect on list yet)
ResourceVersion string
// Timeout for the list/watch call.
TimeoutSeconds *int64
}
// PodLogOptions is the query options for a Pod's logs REST call
......
pkg/api/v1/conversion_generated.go
View file @ 0e1ee14c
......@@ -1127,6 +1127,12 @@ func autoconvert_api_ListOptions_To_v1_ListOptions(in *api.ListOptions, out *Lis
}
out.Watch = in.Watch
out.ResourceVersion = in.ResourceVersion
if in.TimeoutSeconds != nil {
out.TimeoutSeconds = new(int64)
*out.TimeoutSeconds = *in.TimeoutSeconds
} else {
out.TimeoutSeconds = nil
}
return nil
}
......@@ -4146,6 +4152,12 @@ func autoconvert_v1_ListOptions_To_api_ListOptions(in *ListOptions, out *api.Lis
}
out.Watch = in.Watch
out.ResourceVersion = in.ResourceVersion
if in.TimeoutSeconds != nil {
out.TimeoutSeconds = new(int64)
*out.TimeoutSeconds = *in.TimeoutSeconds
} else {
out.TimeoutSeconds = nil
}
return nil
}
......
pkg/api/v1/deep_copy_generated.go
View file @ 0e1ee14c
......@@ -844,6 +844,12 @@ func deepCopy_v1_ListOptions(in ListOptions, out *ListOptions, c *conversion.Clo
out.FieldSelector = in.FieldSelector
out.Watch = in.Watch
out.ResourceVersion = in.ResourceVersion
if in.TimeoutSeconds != nil {
out.TimeoutSeconds = new(int64)
*out.TimeoutSeconds = *in.TimeoutSeconds
} else {
out.TimeoutSeconds = nil
}
return nil
}
......
pkg/api/v1/types.generated.go
View file @ 0e1ee14c
This source diff could not be displayed because it is too large. You can view the blob instead.
pkg/api/v1/types.go
View file @ 0e1ee14c
......@@ -2047,6 +2047,8 @@ type ListOptions struct {
// When specified with a watch call, shows changes that occur after that particular version of a resource.
// Defaults to changes from the beginning of history.
ResourceVersion string `json:"resourceVersion,omitempty"`
// Timeout for the list/watch call.
TimeoutSeconds *int64 `json:"timeoutSeconds,omitempty"`
}
// PodLogOptions is the query options for a Pod's logs REST call.
......
pkg/api/v1/types_swagger_doc_generated.go
View file @ 0e1ee14c
......@@ -546,6 +546,7 @@ var map_ListOptions = map[string]string{
"fieldSelector": "A selector to restrict the list of returned objects by their fields. Defaults to everything.",
"watch": "Watch for changes to the described resources and return them as a stream of add, update, and remove notifications. Specify resourceVersion.",
"resourceVersion": "When specified with a watch call, shows changes that occur after that particular version of a resource. Defaults to changes from the beginning of history.",
"timeoutSeconds": "Timeout for the list/watch call.",
}
func (ListOptions) SwaggerDoc() map[string]string {
......
pkg/apiserver/apiserver_test.go
View file @ 0e1ee14c
......@@ -102,6 +102,7 @@ func addTestTypes() {
FieldSelector string `json:"fields,omitempty"`
Watch bool `json:"watch,omitempty"`
ResourceVersion string `json:"resourceVersion,omitempty"`
TimeoutSeconds *int64 `json:"timeoutSeconds,omitempty"`
}
api.Scheme.AddKnownTypes(
testVersion, &apiservertesting.Simple{}, &apiservertesting.SimpleList{}, &unversioned.Status{},
......@@ -117,6 +118,7 @@ func addNewTestTypes() {
FieldSelector string `json:"fieldSelector,omitempty"`
Watch bool `json:"watch,omitempty"`
ResourceVersion string `json:"resourceVersion,omitempty"`
TimeoutSeconds *int64 `json:"timeoutSeconds,omitempty"`
}
api.Scheme.AddKnownTypes(
newVersion, &apiservertesting.Simple{}, &apiservertesting.SimpleList{}, &unversioned.Status{},
......
pkg/apiserver/resthandler.go
View file @ 0e1ee14c
......@@ -19,6 +19,7 @@ package apiserver
import (
"encoding/json"
"fmt"
"math/rand"
"net/http"
"net/url"
gpath "path"
......@@ -271,7 +272,15 @@ func ListResource(r rest.Lister, rw rest.Watcher, scope RequestScope, forceWatch
errorJSON(err, scope.Codec, w)
return
}
serveWatch(watcher, scope, w, req, minRequestTimeout)
// TODO: Currently we explicitly ignore ?timeout= and use only ?timeoutSeconds=.
timeout := time.Duration(0)
if opts.TimeoutSeconds != nil {
timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
}
if timeout == 0 && minRequestTimeout > 0 {
timeout = time.Duration(float64(minRequestTimeout) * (rand.Float64() + 1.0))
}
serveWatch(watcher, scope, w, req, timeout)
return
}
......
pkg/apiserver/watch.go
View file @ 0e1ee14c
......@@ -17,7 +17,6 @@ limitations under the License.
package apiserver
import (
"math/rand"
"net/http"
"reflect"
"regexp"
......@@ -66,12 +65,7 @@ func (w *realTimeoutFactory) TimeoutCh() (<-chan time.Time, func() bool) {
}
// serveWatch handles serving requests to the server
func serveWatch(watcher watch.Interface, scope RequestScope, w http.ResponseWriter, req *restful.Request, minRequestTimeout time.Duration) {
var timeout time.Duration
if minRequestTimeout > 0 {
// Each watch gets a random timeout between minRequestTimeout and 2*minRequestTimeout to avoid thundering herds.
timeout = time.Duration(float64(minRequestTimeout) * (rand.Float64() + 1.0))
}
func serveWatch(watcher watch.Interface, scope RequestScope, w http.ResponseWriter, req *restful.Request, timeout time.Duration) {
watchServer := &WatchServer{watcher, scope.Codec, func(obj runtime.Object) {
if err := setSelfLink(obj, req, scope.Namer); err != nil {
glog.V(5).Infof("Failed to set self link for object %v: %v", reflect.TypeOf(obj), err)
......
pkg/client/unversioned/daemon_sets.go
View file @ 0e1ee14c
......@@ -17,6 +17,7 @@ limitations under the License.
package unversioned
import (
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/apis/extensions"
"k8s.io/kubernetes/pkg/fields"
"k8s.io/kubernetes/pkg/labels"
......@@ -35,7 +36,7 @@ type DaemonSetInterface interface {
Update(ctrl *extensions.DaemonSet) (*extensions.DaemonSet, error)
UpdateStatus(ctrl *extensions.DaemonSet) (*extensions.DaemonSet, error)
Delete(name string) error
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
}
// daemonSets implements DaemonsSetsNamespacer interface
......@@ -91,12 +92,13 @@ func (c *daemonSets) Delete(name string) error {
}
// Watch returns a watch.Interface that watches the requested daemon sets.
func (c *daemonSets) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *daemonSets) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.r.Get().
Prefix("watch").
Namespace(c.ns).
Resource("daemonsets").
Param("resourceVersion", resourceVersion).
Param("resourceVersion", opts.ResourceVersion).
TimeoutSeconds(TimeoutFromListOptions(opts)).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/deployment.go
View file @ 0e1ee14c
......@@ -36,7 +36,7 @@ type DeploymentInterface interface {
Delete(name string, options *api.DeleteOptions) error
Create(Deployment *extensions.Deployment) (*extensions.Deployment, error)
Update(Deployment *extensions.Deployment) (*extensions.Deployment, error)
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
}
// deployments implements DeploymentInterface
......@@ -94,12 +94,13 @@ func (c *deployments) Update(deployment *extensions.Deployment) (result *extensi
}
// Watch returns a watch.Interface that watches the requested deployments.
func (c *deployments) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *deployments) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.client.Get().
Prefix("watch").
Namespace(c.ns).
Resource("deployments").
Param("resourceVersion", resourceVersion).
Param("resourceVersion", opts.ResourceVersion).
TimeoutSeconds(TimeoutFromListOptions(opts)).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/deployment_test.go
View file @ 0e1ee14c
......@@ -147,6 +147,6 @@ func TestDeploymentWatch(t *testing.T) {
},
Response: Response{StatusCode: 200},
}
_, err := c.Setup(t).Deployments(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), "")
_, err := c.Setup(t).Deployments(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), api.ListOptions{})
c.Validate(t, nil, err)
}
pkg/client/unversioned/endpoints.go
View file @ 0e1ee14c
......@@ -37,7 +37,7 @@ type EndpointsInterface interface {
Get(name string) (*api.Endpoints, error)
Delete(name string) error
Update(endpoints *api.Endpoints) (*api.Endpoints, error)
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
}
// endpoints implements EndpointsInterface
......@@ -83,12 +83,12 @@ func (c *endpoints) Delete(name string) error {
}
// Watch returns a watch.Interface that watches the requested endpoints for a service.
func (c *endpoints) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *endpoints) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.r.Get().
Prefix("watch").
Namespace(c.ns).
Resource("endpoints").
Param("resourceVersion", resourceVersion).
VersionedParams(&opts, api.Scheme).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/events.go
View file @ 0e1ee14c
......@@ -38,7 +38,7 @@ type EventInterface interface {
Patch(event *api.Event, data []byte) (*api.Event, error)
List(label labels.Selector, field fields.Selector) (*api.EventList, error)
Get(name string) (*api.Event, error)
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
// Search finds events about the specified object
Search(objOrRef runtime.Object) (*api.EventList, error)
Delete(name string) error
......@@ -141,12 +141,12 @@ func (e *events) Get(name string) (*api.Event, error) {
}
// Watch starts watching for events matching the given selectors.
func (e *events) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (e *events) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return e.client.Get().
Prefix("watch").
NamespaceIfScoped(e.namespace, len(e.namespace) > 0).
Resource("events").
Param("resourceVersion", resourceVersion).
VersionedParams(&opts, api.Scheme).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/helper.go
View file @ 0e1ee14c
......@@ -631,3 +631,12 @@ func DefaultKubernetesUserAgent() string {
version = seg[0]
return fmt.Sprintf("%s/%s (%s/%s) kubernetes/%s", path.Base(os.Args[0]), version, gruntime.GOOS, gruntime.GOARCH, commit)
}
// TimeoutFromListOptions returns timeout to be set via TimeoutSeconds() method
// based on given options.
func TimeoutFromListOptions(options api.ListOptions) time.Duration {
if options.TimeoutSeconds != nil {
return time.Duration(*options.TimeoutSeconds) * time.Second
}
return 0
}
pkg/client/unversioned/horizontalpodautoscaler.go
View file @ 0e1ee14c
......@@ -37,7 +37,7 @@ type HorizontalPodAutoscalerInterface interface {
Create(horizontalPodAutoscaler *extensions.HorizontalPodAutoscaler) (*extensions.HorizontalPodAutoscaler, error)
Update(horizontalPodAutoscaler *extensions.HorizontalPodAutoscaler) (*extensions.HorizontalPodAutoscaler, error)
UpdateStatus(horizontalPodAutoscaler *extensions.HorizontalPodAutoscaler) (*extensions.HorizontalPodAutoscaler, error)
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
}
// horizontalPodAutoscalers implements HorizontalPodAutoscalersNamespacer interface
......@@ -103,12 +103,13 @@ func (c *horizontalPodAutoscalers) UpdateStatus(horizontalPodAutoscaler *extensi
}
// Watch returns a watch.Interface that watches the requested horizontalPodAutoscalers.
func (c *horizontalPodAutoscalers) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *horizontalPodAutoscalers) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.client.Get().
Prefix("watch").
Namespace(c.ns).
Resource("horizontalPodAutoscalers").
Param("resourceVersion", resourceVersion).
Param("resourceVersion", opts.ResourceVersion).
TimeoutSeconds(TimeoutFromListOptions(opts)).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/horizontalpodautoscaler_test.go
View file @ 0e1ee14c
......@@ -155,6 +155,6 @@ func TestHorizontalPodAutoscalerWatch(t *testing.T) {
Query: url.Values{"resourceVersion": []string{}}},
Response: Response{StatusCode: 200},
}
_, err := c.Setup(t).Extensions().HorizontalPodAutoscalers(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), "")
_, err := c.Setup(t).Extensions().HorizontalPodAutoscalers(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), api.ListOptions{})
c.Validate(t, nil, err)
}
pkg/client/unversioned/ingress.go
View file @ 0e1ee14c
......@@ -36,7 +36,7 @@ type IngressInterface interface {
Create(ingress *extensions.Ingress) (*extensions.Ingress, error)
Update(ingress *extensions.Ingress) (*extensions.Ingress, error)
Delete(name string, options *api.DeleteOptions) error
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
UpdateStatus(ingress *extensions.Ingress) (*extensions.Ingress, error)
}
......@@ -93,12 +93,13 @@ func (c *ingress) Delete(name string, options *api.DeleteOptions) (err error) {
}
// Watch returns a watch.Interface that watches the requested ingress.
func (c *ingress) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *ingress) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.r.Get().
Prefix("watch").
Namespace(c.ns).
Resource("ingress").
Param("resourceVersion", resourceVersion).
Param("resourceVersion", opts.ResourceVersion).
TimeoutSeconds(TimeoutFromListOptions(opts)).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/jobs.go
View file @ 0e1ee14c
......@@ -37,7 +37,7 @@ type JobInterface interface {
Create(job *extensions.Job) (*extensions.Job, error)
Update(job *extensions.Job) (*extensions.Job, error)
Delete(name string, options *api.DeleteOptions) error
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
UpdateStatus(job *extensions.Job) (*extensions.Job, error)
}
......@@ -97,12 +97,13 @@ func (c *jobs) Delete(name string, options *api.DeleteOptions) (err error) {
}
// Watch returns a watch.Interface that watches the requested jobs.
func (c *jobs) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *jobs) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.r.Get().
Prefix("watch").
Namespace(c.ns).
Resource("jobs").
Param("resourceVersion", resourceVersion).
Param("resourceVersion", opts.ResourceVersion).
TimeoutSeconds(TimeoutFromListOptions(opts)).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/limit_ranges.go
View file @ 0e1ee14c
......@@ -37,7 +37,7 @@ type LimitRangeInterface interface {
Delete(name string) error
Create(limitRange *api.LimitRange) (*api.LimitRange, error)
Update(limitRange *api.LimitRange) (*api.LimitRange, error)
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
}
// limitRanges implements LimitRangesNamespacer interface
......@@ -92,12 +92,12 @@ func (c *limitRanges) Update(limitRange *api.LimitRange) (result *api.LimitRange
}
// Watch returns a watch.Interface that watches the requested resource
func (c *limitRanges) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *limitRanges) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.r.Get().
Prefix("watch").
Namespace(c.ns).
Resource("limitRanges").
Param("resourceVersion", resourceVersion).
VersionedParams(&opts, api.Scheme).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/limit_ranges_test.go
View file @ 0e1ee14c
......@@ -207,6 +207,6 @@ func TestLimitRangeWatch(t *testing.T) {
Query: url.Values{"resourceVersion": []string{}}},
Response: Response{StatusCode: 200},
}
_, err := c.Setup(t).LimitRanges(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), "")
_, err := c.Setup(t).LimitRanges(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), api.ListOptions{})
c.Validate(t, nil, err)
}
pkg/client/unversioned/namespaces.go
View file @ 0e1ee14c
......@@ -35,7 +35,7 @@ type NamespaceInterface interface {
List(label labels.Selector, field fields.Selector) (*api.NamespaceList, error)
Delete(name string) error
Update(item *api.Namespace) (*api.Namespace, error)
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
Finalize(item *api.Namespace) (*api.Namespace, error)
Status(item *api.Namespace) (*api.Namespace, error)
}
......@@ -114,11 +114,11 @@ func (c *namespaces) Delete(name string) error {
}
// Watch returns a watch.Interface that watches the requested namespaces.
func (c *namespaces) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *namespaces) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.r.Get().
Prefix("watch").
Resource("namespaces").
Param("resourceVersion", resourceVersion).
VersionedParams(&opts, api.Scheme).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/namespaces_test.go
View file @ 0e1ee14c
......@@ -174,6 +174,6 @@ func TestNamespaceWatch(t *testing.T) {
Query: url.Values{"resourceVersion": []string{}}},
Response: Response{StatusCode: 200},
}
_, err := c.Setup(t).Namespaces().Watch(labels.Everything(), fields.Everything(), "")
_, err := c.Setup(t).Namespaces().Watch(labels.Everything(), fields.Everything(), api.ListOptions{})
c.Validate(t, nil, err)
}
pkg/client/unversioned/nodes.go
View file @ 0e1ee14c
......@@ -36,7 +36,7 @@ type NodeInterface interface {
Delete(name string) error
Update(*api.Node) (*api.Node, error)
UpdateStatus(*api.Node) (*api.Node, error)
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
}
// nodes implements NodesInterface
......@@ -102,12 +102,12 @@ func (c *nodes) UpdateStatus(node *api.Node) (*api.Node, error) {
}
// Watch returns a watch.Interface that watches the requested nodes.
func (c *nodes) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *nodes) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.r.Get().
Prefix("watch").
Namespace(api.NamespaceAll).
Resource(c.resourceName()).
Param("resourceVersion", resourceVersion).
VersionedParams(&opts, api.Scheme).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/persistentvolume_test.go
View file @ 0e1ee14c
......@@ -180,6 +180,6 @@ func TestPersistentVolumeWatch(t *testing.T) {
Query: url.Values{"resourceVersion": []string{}}},
Response: Response{StatusCode: 200},
}
_, err := c.Setup(t).PersistentVolumes().Watch(labels.Everything(), fields.Everything(), "")
_, err := c.Setup(t).PersistentVolumes().Watch(labels.Everything(), fields.Everything(), api.ListOptions{})
c.Validate(t, nil, err)
}
pkg/client/unversioned/persistentvolumeclaim.go
View file @ 0e1ee14c
......@@ -38,7 +38,7 @@ type PersistentVolumeClaimInterface interface {
Update(claim *api.PersistentVolumeClaim) (*api.PersistentVolumeClaim, error)
UpdateStatus(claim *api.PersistentVolumeClaim) (*api.PersistentVolumeClaim, error)
Delete(name string) error
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
}
// persistentVolumeClaims implements PersistentVolumeClaimsNamespacer interface
......@@ -98,12 +98,12 @@ func (c *persistentVolumeClaims) Delete(name string) error {
return c.client.Delete().Namespace(c.namespace).Resource("persistentVolumeClaims").Name(name).Do().Error()
}
func (c *persistentVolumeClaims) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *persistentVolumeClaims) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.client.Get().
Prefix("watch").
Namespace(c.namespace).
Resource("persistentVolumeClaims").
Param("resourceVersion", resourceVersion).
VersionedParams(&opts, api.Scheme).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/persistentvolumeclaim_test.go
View file @ 0e1ee14c
......@@ -197,6 +197,6 @@ func TestPersistentVolumeClaimWatch(t *testing.T) {
Query: url.Values{"resourceVersion": []string{}}},
Response: Response{StatusCode: 200},
}
_, err := c.Setup(t).PersistentVolumeClaims(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), "")
_, err := c.Setup(t).PersistentVolumeClaims(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), api.ListOptions{})
c.Validate(t, nil, err)
}
pkg/client/unversioned/persistentvolumes.go
View file @ 0e1ee14c
......@@ -37,7 +37,7 @@ type PersistentVolumeInterface interface {
Update(volume *api.PersistentVolume) (*api.PersistentVolume, error)
UpdateStatus(persistentVolume *api.PersistentVolume) (*api.PersistentVolume, error)
Delete(name string) error
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
}
// persistentVolumes implements PersistentVolumesInterface
......@@ -93,11 +93,11 @@ func (c *persistentVolumes) Delete(name string) error {
return c.client.Delete().Resource("persistentVolumes").Name(name).Do().Error()
}
func (c *persistentVolumes) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *persistentVolumes) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.client.Get().
Prefix("watch").
Resource("persistentVolumes").
Param("resourceVersion", resourceVersion).
VersionedParams(&opts, api.Scheme).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/pod_templates.go
View file @ 0e1ee14c
......@@ -35,7 +35,7 @@ type PodTemplateInterface interface {
Delete(name string, options *api.DeleteOptions) error
Create(podTemplate *api.PodTemplate) (*api.PodTemplate, error)
Update(podTemplate *api.PodTemplate) (*api.PodTemplate, error)
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
}
// podTemplates implements PodTemplatesNamespacer interface
......@@ -94,12 +94,12 @@ func (c *podTemplates) Update(podTemplate *api.PodTemplate) (result *api.PodTemp
}
// Watch returns a watch.Interface that watches the requested podTemplates.
func (c *podTemplates) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *podTemplates) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.r.Get().
Prefix("watch").
Namespace(c.ns).
Resource("podTemplates").
Param("resourceVersion", resourceVersion).
VersionedParams(&opts, api.Scheme).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/pod_templates_test.go
View file @ 0e1ee14c
......@@ -137,6 +137,6 @@ func TestPodTemplateWatch(t *testing.T) {
Query: url.Values{"resourceVersion": []string{}}},
Response: Response{StatusCode: 200},
}
_, err := c.Setup(t).PodTemplates(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), "")
_, err := c.Setup(t).PodTemplates(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), api.ListOptions{})
c.Validate(t, nil, err)
}
pkg/client/unversioned/pods.go
View file @ 0e1ee14c
......@@ -35,7 +35,7 @@ type PodInterface interface {
Delete(name string, options *api.DeleteOptions) error
Create(pod *api.Pod) (*api.Pod, error)
Update(pod *api.Pod) (*api.Pod, error)
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
Bind(binding *api.Binding) error
UpdateStatus(pod *api.Pod) (*api.Pod, error)
}
......@@ -96,12 +96,12 @@ func (c *pods) Update(pod *api.Pod) (result *api.Pod, err error) {
}
// Watch returns a watch.Interface that watches the requested pods.
func (c *pods) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *pods) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.r.Get().
Prefix("watch").
Namespace(c.ns).
Resource("pods").
Param("resourceVersion", resourceVersion).
VersionedParams(&opts, api.Scheme).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/replication_controllers.go
View file @ 0e1ee14c
......@@ -36,7 +36,7 @@ type ReplicationControllerInterface interface {
Update(ctrl *api.ReplicationController) (*api.ReplicationController, error)
UpdateStatus(ctrl *api.ReplicationController) (*api.ReplicationController, error)
Delete(name string) error
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
}
// replicationControllers implements ReplicationControllersNamespacer interface
......@@ -91,12 +91,12 @@ func (c *replicationControllers) Delete(name string) error {
}
// Watch returns a watch.Interface that watches the requested controllers.
func (c *replicationControllers) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *replicationControllers) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.r.Get().
Prefix("watch").
Namespace(c.ns).
Resource("replicationControllers").
Param("resourceVersion", resourceVersion).
VersionedParams(&opts, api.Scheme).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/request.go
View file @ 0e1ee14c
......@@ -459,6 +459,18 @@ func (r *Request) Timeout(d time.Duration) *Request {
return r
}
// Timeout makes the request use the given duration as a timeout. Sets the "timeoutSeconds"
// parameter.
func (r *Request) TimeoutSeconds(d time.Duration) *Request {
if r.err != nil {
return r
}
if d != 0 {
r.Param("timeoutSeconds", d.String())
}
return r
}
// Body makes the request use obj as the body. Optional.
// If obj is a string, try to read a file of that name.
// If obj is a []byte, send it directly.
......
pkg/client/unversioned/request_test.go
View file @ 0e1ee14c
......@@ -173,6 +173,25 @@ func TestRequestVersionedParams(t *testing.T) {
}
}
func TestRequestVersionedParamsFromListOptions(t *testing.T) {
r := &Request{apiVersion: "v1"}
r.VersionedParams(&api.ListOptions{ResourceVersion: "1"}, api.Scheme)
if !reflect.DeepEqual(r.params, url.Values{
"resourceVersion": []string{"1"},
}) {
t.Errorf("should have set a param: %#v", r)
}
var timeout int64 = 10
r.VersionedParams(&api.ListOptions{ResourceVersion: "2", TimeoutSeconds: &timeout}, api.Scheme)
if !reflect.DeepEqual(r.params, url.Values{
"resourceVersion": []string{"1", "2"},
"timeoutSeconds": []string{"10"},
}) {
t.Errorf("should have set a param: %#v", r)
}
}
func TestRequestURI(t *testing.T) {
r := (&Request{}).Param("foo", "a")
r.Prefix("other")
......
pkg/client/unversioned/resource_quotas.go
View file @ 0e1ee14c
......@@ -36,7 +36,7 @@ type ResourceQuotaInterface interface {
Create(resourceQuota *api.ResourceQuota) (*api.ResourceQuota, error)
Update(resourceQuota *api.ResourceQuota) (*api.ResourceQuota, error)
UpdateStatus(resourceQuota *api.ResourceQuota) (*api.ResourceQuota, error)
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
}
// resourceQuotas implements ResourceQuotasNamespacer interface
......@@ -94,12 +94,12 @@ func (c *resourceQuotas) UpdateStatus(resourceQuota *api.ResourceQuota) (result
}
// Watch returns a watch.Interface that watches the requested resource
func (c *resourceQuotas) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *resourceQuotas) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.r.Get().
Prefix("watch").
Namespace(c.ns).
Resource("resourceQuotas").
Param("resourceVersion", resourceVersion).
VersionedParams(&opts, api.Scheme).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/resource_quotas_test.go
View file @ 0e1ee14c
......@@ -193,6 +193,6 @@ func TestResourceQuotaWatch(t *testing.T) {
Query: url.Values{"resourceVersion": []string{}}},
Response: Response{StatusCode: 200},
}
_, err := c.Setup(t).ResourceQuotas(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), "")
_, err := c.Setup(t).ResourceQuotas(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), api.ListOptions{})
c.Validate(t, nil, err)
}
pkg/client/unversioned/secrets.go
View file @ 0e1ee14c
......@@ -33,7 +33,7 @@ type SecretsInterface interface {
Delete(name string) error
List(label labels.Selector, field fields.Selector) (*api.SecretList, error)
Get(name string) (*api.Secret, error)
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
}
// events implements Secrets interface
......@@ -91,12 +91,12 @@ func (s *secrets) Get(name string) (*api.Secret, error) {
}
// Watch starts watching for secrets matching the given selectors.
func (s *secrets) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (s *secrets) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return s.client.Get().
Prefix("watch").
Namespace(s.namespace).
Resource("secrets").
Param("resourceVersion", resourceVersion).
VersionedParams(&opts, api.Scheme).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/service_accounts.go
View file @ 0e1ee14c
......@@ -33,7 +33,7 @@ type ServiceAccountsInterface interface {
Delete(name string) error
List(label labels.Selector, field fields.Selector) (*api.ServiceAccountList, error)
Get(name string) (*api.ServiceAccount, error)
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
}
// serviceAccounts implements ServiceAccounts interface
......@@ -91,12 +91,12 @@ func (s *serviceAccounts) Get(name string) (*api.ServiceAccount, error) {
}
// Watch starts watching for serviceAccounts matching the given selectors.
func (s *serviceAccounts) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (s *serviceAccounts) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return s.client.Get().
Prefix("watch").
Namespace(s.namespace).
Resource("serviceAccounts").
Param("resourceVersion", resourceVersion).
VersionedParams(&opts, api.Scheme).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/services.go
View file @ 0e1ee14c
......@@ -35,7 +35,7 @@ type ServiceInterface interface {
Create(srv *api.Service) (*api.Service, error)
Update(srv *api.Service) (*api.Service, error)
Delete(name string) error
Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error)
ProxyGet(name, path string, params map[string]string) ResponseWrapper
}
......@@ -90,12 +90,12 @@ func (c *services) Delete(name string) error {
}
// Watch returns a watch.Interface that watches the requested services.
func (c *services) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (c *services) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.r.Get().
Prefix("watch").
Namespace(c.ns).
Resource("services").
Param("resourceVersion", resourceVersion).
VersionedParams(&opts, api.Scheme).
LabelsSelectorParam(label).
FieldsSelectorParam(field).
Watch()
......
pkg/client/unversioned/testclient/actions.go
View file @ 0e1ee14c
......@@ -19,6 +19,7 @@ package testclient
import (
"strings"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/fields"
"k8s.io/kubernetes/pkg/labels"
"k8s.io/kubernetes/pkg/runtime"
......@@ -149,21 +150,21 @@ func NewDeleteAction(resource, namespace, name string) DeleteActionImpl {
return action
}
func NewRootWatchAction(resource string, label labels.Selector, field fields.Selector, resourceVersion string) WatchActionImpl {
func NewRootWatchAction(resource string, label labels.Selector, field fields.Selector, opts api.ListOptions) WatchActionImpl {
action := WatchActionImpl{}
action.Verb = "watch"
action.Resource = resource
action.WatchRestrictions = WatchRestrictions{label, field, resourceVersion}
action.WatchRestrictions = WatchRestrictions{label, field, opts.ResourceVersion}
return action
}
func NewWatchAction(resource, namespace string, label labels.Selector, field fields.Selector, resourceVersion string) WatchActionImpl {
func NewWatchAction(resource, namespace string, label labels.Selector, field fields.Selector, opts api.ListOptions) WatchActionImpl {
action := WatchActionImpl{}
action.Verb = "watch"
action.Resource = resource
action.Namespace = namespace
action.WatchRestrictions = WatchRestrictions{label, field, resourceVersion}
action.WatchRestrictions = WatchRestrictions{label, field, opts.ResourceVersion}
return action
}
......
pkg/client/unversioned/testclient/fake_daemon_sets.go
View file @ 0e1ee14c
......@@ -17,6 +17,7 @@ limitations under the License.
package testclient
import (
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/apis/extensions"
kclientlib "k8s.io/kubernetes/pkg/client/unversioned"
"k8s.io/kubernetes/pkg/fields"
......@@ -79,6 +80,6 @@ func (c *FakeDaemonSets) Delete(name string) error {
return err
}
func (c *FakeDaemonSets) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("daemonsets", c.Namespace, label, field, resourceVersion))
func (c *FakeDaemonSets) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("daemonsets", c.Namespace, label, field, opts))
}
pkg/client/unversioned/testclient/fake_deployments.go
View file @ 0e1ee14c
......@@ -77,6 +77,6 @@ func (c *FakeDeployments) Delete(name string, options *api.DeleteOptions) error
return err
}
func (c *FakeDeployments) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("deployments", c.Namespace, label, field, resourceVersion))
func (c *FakeDeployments) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("deployments", c.Namespace, label, field, opts))
}
pkg/client/unversioned/testclient/fake_endpoints.go
View file @ 0e1ee14c
......@@ -71,6 +71,6 @@ func (c *FakeEndpoints) Delete(name string) error {
return err
}
func (c *FakeEndpoints) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("endpoints", c.Namespace, label, field, resourceVersion))
func (c *FakeEndpoints) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("endpoints", c.Namespace, label, field, opts))
}
pkg/client/unversioned/testclient/fake_events.go
View file @ 0e1ee14c
......@@ -111,10 +111,10 @@ func (c *FakeEvents) Delete(name string) error {
}
// Watch starts watching for events matching the given selectors.
func (c *FakeEvents) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
action := NewRootWatchAction("events", label, field, resourceVersion)
func (c *FakeEvents) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
action := NewRootWatchAction("events", label, field, opts)
if c.Namespace != "" {
action = NewWatchAction("events", c.Namespace, label, field, resourceVersion)
action = NewWatchAction("events", c.Namespace, label, field, opts)
}
return c.Fake.InvokesWatch(action)
}
......
pkg/client/unversioned/testclient/fake_horizontal_pod_autoscalers.go
View file @ 0e1ee14c
......@@ -85,6 +85,6 @@ func (c *FakeHorizontalPodAutoscalers) Delete(name string, options *api.DeleteOp
return err
}
func (c *FakeHorizontalPodAutoscalers) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("horizontalpodautoscalers", c.Namespace, label, field, resourceVersion))
func (c *FakeHorizontalPodAutoscalers) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("horizontalpodautoscalers", c.Namespace, label, field, opts))
}
pkg/client/unversioned/testclient/fake_ingress.go
View file @ 0e1ee14c
......@@ -72,8 +72,8 @@ func (c *FakeIngress) Delete(name string, options *api.DeleteOptions) error {
return err
}
func (c *FakeIngress) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("ingress", c.Namespace, label, field, resourceVersion))
func (c *FakeIngress) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("ingress", c.Namespace, label, field, opts))
}
func (c *FakeIngress) UpdateStatus(ingress *extensions.Ingress) (result *extensions.Ingress, err error) {
......
pkg/client/unversioned/testclient/fake_jobs.go
View file @ 0e1ee14c
......@@ -72,8 +72,8 @@ func (c *FakeJobs) Delete(name string, options *api.DeleteOptions) error {
return err
}
func (c *FakeJobs) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("jobs", c.Namespace, label, field, resourceVersion))
func (c *FakeJobs) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("jobs", c.Namespace, label, field, opts))
}
func (c *FakeJobs) UpdateStatus(job *extensions.Job) (result *extensions.Job, err error) {
......
pkg/client/unversioned/testclient/fake_limit_ranges.go
View file @ 0e1ee14c
......@@ -71,6 +71,6 @@ func (c *FakeLimitRanges) Delete(name string) error {
return err
}
func (c *FakeLimitRanges) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("limitranges", c.Namespace, label, field, resourceVersion))
func (c *FakeLimitRanges) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("limitranges", c.Namespace, label, field, opts))
}
pkg/client/unversioned/testclient/fake_namespaces.go
View file @ 0e1ee14c
......@@ -70,8 +70,8 @@ func (c *FakeNamespaces) Delete(name string) error {
return err
}
func (c *FakeNamespaces) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewRootWatchAction("namespaces", label, field, resourceVersion))
func (c *FakeNamespaces) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewRootWatchAction("namespaces", label, field, opts))
}
func (c *FakeNamespaces) Finalize(namespace *api.Namespace) (*api.Namespace, error) {
......
pkg/client/unversioned/testclient/fake_nodes.go
View file @ 0e1ee14c
......@@ -70,8 +70,8 @@ func (c *FakeNodes) Delete(name string) error {
return err
}
func (c *FakeNodes) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewRootWatchAction("nodes", label, field, resourceVersion))
func (c *FakeNodes) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewRootWatchAction("nodes", label, field, opts))
}
func (c *FakeNodes) UpdateStatus(node *api.Node) (*api.Node, error) {
......
pkg/client/unversioned/testclient/fake_persistent_volume_claims.go
View file @ 0e1ee14c
......@@ -69,8 +69,8 @@ func (c *FakePersistentVolumeClaims) Delete(name string) error {
return err
}
func (c *FakePersistentVolumeClaims) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("persistentvolumeclaims", c.Namespace, label, field, resourceVersion))
func (c *FakePersistentVolumeClaims) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("persistentvolumeclaims", c.Namespace, label, field, opts))
}
func (c *FakePersistentVolumeClaims) UpdateStatus(claim *api.PersistentVolumeClaim) (*api.PersistentVolumeClaim, error) {
......
pkg/client/unversioned/testclient/fake_persistent_volumes.go
View file @ 0e1ee14c
......@@ -68,8 +68,8 @@ func (c *FakePersistentVolumes) Delete(name string) error {
return err
}
func (c *FakePersistentVolumes) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewRootWatchAction("persistentvolumes", label, field, resourceVersion))
func (c *FakePersistentVolumes) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewRootWatchAction("persistentvolumes", label, field, opts))
}
func (c *FakePersistentVolumes) UpdateStatus(pv *api.PersistentVolume) (*api.PersistentVolume, error) {
......
pkg/client/unversioned/testclient/fake_pod_templates.go
View file @ 0e1ee14c
......@@ -71,6 +71,6 @@ func (c *FakePodTemplates) Delete(name string, options *api.DeleteOptions) error
return err
}
func (c *FakePodTemplates) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("podtemplates", c.Namespace, label, field, resourceVersion))
func (c *FakePodTemplates) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("podtemplates", c.Namespace, label, field, opts))
}
pkg/client/unversioned/testclient/fake_pods.go
View file @ 0e1ee14c
......@@ -76,8 +76,8 @@ func (c *FakePods) Delete(name string, options *api.DeleteOptions) error {
return err
}
func (c *FakePods) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("pods", c.Namespace, label, field, resourceVersion))
func (c *FakePods) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("pods", c.Namespace, label, field, opts))
}
func (c *FakePods) Bind(binding *api.Binding) error {
......
pkg/client/unversioned/testclient/fake_replication_controllers.go
View file @ 0e1ee14c
......@@ -79,6 +79,6 @@ func (c *FakeReplicationControllers) Delete(name string) error {
return err
}
func (c *FakeReplicationControllers) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("replicationcontrollers", c.Namespace, label, field, resourceVersion))
func (c *FakeReplicationControllers) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("replicationcontrollers", c.Namespace, label, field, opts))
}
pkg/client/unversioned/testclient/fake_resource_quotas.go
View file @ 0e1ee14c
......@@ -71,8 +71,8 @@ func (c *FakeResourceQuotas) Delete(name string) error {
return err
}
func (c *FakeResourceQuotas) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("resourcequotas", c.Namespace, label, field, resourceVersion))
func (c *FakeResourceQuotas) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("resourcequotas", c.Namespace, label, field, opts))
}
func (c *FakeResourceQuotas) UpdateStatus(resourceQuota *api.ResourceQuota) (*api.ResourceQuota, error) {
......
pkg/client/unversioned/testclient/fake_secrets.go
View file @ 0e1ee14c
......@@ -71,6 +71,6 @@ func (c *FakeSecrets) Delete(name string) error {
return err
}
func (c *FakeSecrets) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("secrets", c.Namespace, label, field, resourceVersion))
func (c *FakeSecrets) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("secrets", c.Namespace, label, field, opts))
}
pkg/client/unversioned/testclient/fake_service_accounts.go
View file @ 0e1ee14c
......@@ -71,6 +71,6 @@ func (c *FakeServiceAccounts) Delete(name string) error {
return err
}
func (c *FakeServiceAccounts) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("serviceaccounts", c.Namespace, label, field, resourceVersion))
func (c *FakeServiceAccounts) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("serviceaccounts", c.Namespace, label, field, opts))
}
pkg/client/unversioned/testclient/fake_services.go
View file @ 0e1ee14c
......@@ -72,8 +72,8 @@ func (c *FakeServices) Delete(name string) error {
return err
}
func (c *FakeServices) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("services", c.Namespace, label, field, resourceVersion))
func (c *FakeServices) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return c.Fake.InvokesWatch(NewWatchAction("services", c.Namespace, label, field, opts))
}
func (c *FakeServices) ProxyGet(name, path string, params map[string]string) unversioned.ResponseWrapper {
......
pkg/controller/daemon/controller.go
View file @ 0e1ee14c
......@@ -100,7 +100,8 @@ func NewDaemonSetsController(kubeClient client.Interface, resyncPeriod controlle
return dsc.kubeClient.Extensions().DaemonSets(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return dsc.kubeClient.Extensions().DaemonSets(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return dsc.kubeClient.Extensions().DaemonSets(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
},
&extensions.DaemonSet{},
......@@ -132,7 +133,8 @@ func NewDaemonSetsController(kubeClient client.Interface, resyncPeriod controlle
return dsc.kubeClient.Pods(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return dsc.kubeClient.Pods(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return dsc.kubeClient.Pods(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Pod{},
......@@ -150,7 +152,8 @@ func NewDaemonSetsController(kubeClient client.Interface, resyncPeriod controlle
return dsc.kubeClient.Nodes().List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return dsc.kubeClient.Nodes().Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return dsc.kubeClient.Nodes().Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Node{},
......
pkg/controller/endpoint/endpoints_controller.go
View file @ 0e1ee14c
......@@ -65,7 +65,8 @@ func NewEndpointController(client *client.Client, resyncPeriod controller.Resync
return e.client.Services(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return e.client.Services(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return e.client.Services(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Service{},
......@@ -86,7 +87,8 @@ func NewEndpointController(client *client.Client, resyncPeriod controller.Resync
return e.client.Pods(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return e.client.Pods(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return e.client.Pods(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Pod{},
......
pkg/controller/gc/gc_controller.go
View file @ 0e1ee14c
......@@ -69,7 +69,8 @@ func New(kubeClient client.Interface, resyncPeriod controller.ResyncPeriodFunc,
return gcc.kubeClient.Pods(api.NamespaceAll).List(labels.Everything(), terminatedSelector)
},
WatchFunc: func(rv string) (watch.Interface, error) {
return gcc.kubeClient.Pods(api.NamespaceAll).Watch(labels.Everything(), terminatedSelector, rv)
options := api.ListOptions{ResourceVersion: rv}
return gcc.kubeClient.Pods(api.NamespaceAll).Watch(labels.Everything(), terminatedSelector, options)
},
},
&api.Pod{},
......
pkg/controller/job/controller.go
View file @ 0e1ee14c
......@@ -89,7 +89,8 @@ func NewJobController(kubeClient client.Interface, resyncPeriod controller.Resyn
return jm.kubeClient.Extensions().Jobs(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return jm.kubeClient.Extensions().Jobs(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return jm.kubeClient.Extensions().Jobs(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
},
&extensions.Job{},
......@@ -112,7 +113,8 @@ func NewJobController(kubeClient client.Interface, resyncPeriod controller.Resyn
return jm.kubeClient.Pods(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return jm.kubeClient.Pods(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return jm.kubeClient.Pods(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Pod{},
......
pkg/controller/namespace/namespace_controller.go
View file @ 0e1ee14c
......@@ -51,7 +51,8 @@ func NewNamespaceController(kubeClient client.Interface, versions *unversioned.A
return kubeClient.Namespaces().List(labels.Everything(), fields.Everything())
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return kubeClient.Namespaces().Watch(labels.Everything(), fields.Everything(), resourceVersion)
options := api.ListOptions{ResourceVersion: resourceVersion}
return kubeClient.Namespaces().Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Namespace{},
......
pkg/controller/node/nodecontroller.go
View file @ 0e1ee14c
......@@ -167,7 +167,8 @@ func NewNodeController(
return nc.kubeClient.Pods(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return nc.kubeClient.Pods(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return nc.kubeClient.Pods(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Pod{},
......@@ -183,7 +184,8 @@ func NewNodeController(
return nc.kubeClient.Nodes().List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return nc.kubeClient.Nodes().Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return nc.kubeClient.Nodes().Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Node{},
......
pkg/controller/node/nodecontroller_test.go
View file @ 0e1ee14c
......@@ -137,7 +137,7 @@ func (m *FakeNodeHandler) UpdateStatus(node *api.Node) (*api.Node, error) {
return node, nil
}
func (m *FakeNodeHandler) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
func (m *FakeNodeHandler) Watch(label labels.Selector, field fields.Selector, opts api.ListOptions) (watch.Interface, error) {
return nil, nil
}
......
pkg/controller/persistentvolume/persistentvolume_claim_binder_controller.go
View file @ 0e1ee14c
......@@ -59,7 +59,8 @@ func NewPersistentVolumeClaimBinder(kubeClient client.Interface, syncPeriod time
return kubeClient.PersistentVolumes().List(labels.Everything(), fields.Everything())
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return kubeClient.PersistentVolumes().Watch(labels.Everything(), fields.Everything(), resourceVersion)
options := api.ListOptions{ResourceVersion: resourceVersion}
return kubeClient.PersistentVolumes().Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.PersistentVolume{},
......@@ -77,7 +78,8 @@ func NewPersistentVolumeClaimBinder(kubeClient client.Interface, syncPeriod time
return kubeClient.PersistentVolumeClaims(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return kubeClient.PersistentVolumeClaims(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), resourceVersion)
options := api.ListOptions{ResourceVersion: resourceVersion}
return kubeClient.PersistentVolumeClaims(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.PersistentVolumeClaim{},
......
pkg/controller/persistentvolume/persistentvolume_recycler_controller.go
View file @ 0e1ee14c
......@@ -67,7 +67,8 @@ func NewPersistentVolumeRecycler(kubeClient client.Interface, syncPeriod time.Du
return kubeClient.PersistentVolumes().List(labels.Everything(), fields.Everything())
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return kubeClient.PersistentVolumes().Watch(labels.Everything(), fields.Everything(), resourceVersion)
options := api.ListOptions{ResourceVersion: resourceVersion}
return kubeClient.PersistentVolumes().Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.PersistentVolume{},
......
pkg/controller/replication/replication_controller.go
View file @ 0e1ee14c
......@@ -111,7 +111,8 @@ func NewReplicationManager(kubeClient client.Interface, resyncPeriod controller.
return rm.kubeClient.ReplicationControllers(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return rm.kubeClient.ReplicationControllers(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return rm.kubeClient.ReplicationControllers(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.ReplicationController{},
......@@ -152,7 +153,8 @@ func NewReplicationManager(kubeClient client.Interface, resyncPeriod controller.
return rm.kubeClient.Pods(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return rm.kubeClient.Pods(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return rm.kubeClient.Pods(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Pod{},
......
pkg/controller/serviceaccount/serviceaccounts_controller.go
View file @ 0e1ee14c
......@@ -80,7 +80,8 @@ func NewServiceAccountsController(cl client.Interface, options ServiceAccountsCo
return e.client.ServiceAccounts(api.NamespaceAll).List(labels.Everything(), accountSelector)
},
WatchFunc: func(rv string) (watch.Interface, error) {
return e.client.ServiceAccounts(api.NamespaceAll).Watch(labels.Everything(), accountSelector, rv)
options := api.ListOptions{ResourceVersion: rv}
return e.client.ServiceAccounts(api.NamespaceAll).Watch(labels.Everything(), accountSelector, options)
},
},
&api.ServiceAccount{},
......@@ -97,7 +98,8 @@ func NewServiceAccountsController(cl client.Interface, options ServiceAccountsCo
return e.client.Namespaces().List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return e.client.Namespaces().Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return e.client.Namespaces().Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Namespace{},
......
pkg/controller/serviceaccount/tokens_controller.go
View file @ 0e1ee14c
......@@ -66,7 +66,8 @@ func NewTokensController(cl client.Interface, options TokensControllerOptions) *
return e.client.ServiceAccounts(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return e.client.ServiceAccounts(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return e.client.ServiceAccounts(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.ServiceAccount{},
......@@ -86,7 +87,8 @@ func NewTokensController(cl client.Interface, options TokensControllerOptions) *
return e.client.Secrets(api.NamespaceAll).List(labels.Everything(), tokenSelector)
},
WatchFunc: func(rv string) (watch.Interface, error) {
return e.client.Secrets(api.NamespaceAll).Watch(labels.Everything(), tokenSelector, rv)
options := api.ListOptions{ResourceVersion: rv}
return e.client.Secrets(api.NamespaceAll).Watch(labels.Everything(), tokenSelector, options)
},
},
&api.Secret{},
......
pkg/kubelet/kubelet.go
View file @ 0e1ee14c
......@@ -214,7 +214,8 @@ func NewMainKubelet(
return kubeClient.Services(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return kubeClient.Services(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), resourceVersion)
options := api.ListOptions{ResourceVersion: resourceVersion}
return kubeClient.Services(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
}
cache.NewReflector(listWatch, &api.Service{}, serviceStore, 0).Run()
......@@ -231,7 +232,8 @@ func NewMainKubelet(
return kubeClient.Nodes().List(labels.Everything(), fieldSelector)
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return kubeClient.Nodes().Watch(labels.Everything(), fieldSelector, resourceVersion)
options := api.ListOptions{ResourceVersion: resourceVersion}
return kubeClient.Nodes().Watch(labels.Everything(), fieldSelector, options)
},
}
cache.NewReflector(listWatch, &api.Node{}, nodeStore, 0).Run()
......
pkg/volume/util.go
View file @ 0e1ee14c
......@@ -112,7 +112,8 @@ func (c *realRecyclerClient) WatchPod(name, namespace, resourceVersion string, s
return c.client.Pods(namespace).List(labels.Everything(), fieldSelector)
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return c.client.Pods(namespace).Watch(labels.Everything(), fieldSelector, resourceVersion)
options := api.ListOptions{ResourceVersion: resourceVersion}
return c.client.Pods(namespace).Watch(labels.Everything(), fieldSelector, options)
},
}
queue := cache.NewFIFO(cache.MetaNamespaceKeyFunc)
......
plugin/pkg/admission/limitranger/admission.go
View file @ 0e1ee14c
......@@ -103,7 +103,8 @@ func NewLimitRanger(client client.Interface, limitFunc LimitFunc) admission.Inte
return client.LimitRanges(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return client.LimitRanges(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), resourceVersion)
options := api.ListOptions{ResourceVersion: resourceVersion}
return client.LimitRanges(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
}
indexer, reflector := cache.NewNamespaceKeyedIndexerAndReflector(lw, &api.LimitRange{}, 0)
......
plugin/pkg/admission/namespace/autoprovision/admission.go
View file @ 0e1ee14c
......@@ -88,7 +88,8 @@ func NewProvision(c client.Interface) admission.Interface {
return c.Namespaces().List(labels.Everything(), fields.Everything())
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return c.Namespaces().Watch(labels.Everything(), fields.Everything(), resourceVersion)
options := api.ListOptions{ResourceVersion: resourceVersion}
return c.Namespaces().Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Namespace{},
......
plugin/pkg/admission/namespace/exists/admission.go
View file @ 0e1ee14c
......@@ -95,7 +95,8 @@ func NewExists(c client.Interface) admission.Interface {
return c.Namespaces().List(labels.Everything(), fields.Everything())
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return c.Namespaces().Watch(labels.Everything(), fields.Everything(), resourceVersion)
options := api.ListOptions{ResourceVersion: resourceVersion}
return c.Namespaces().Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Namespace{},
......
plugin/pkg/admission/namespace/lifecycle/admission.go
View file @ 0e1ee14c
......@@ -112,7 +112,8 @@ func NewLifecycle(c client.Interface) admission.Interface {
return c.Namespaces().List(labels.Everything(), fields.Everything())
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return c.Namespaces().Watch(labels.Everything(), fields.Everything(), resourceVersion)
options := api.ListOptions{ResourceVersion: resourceVersion}
return c.Namespaces().Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Namespace{},
......
plugin/pkg/admission/resourcequota/admission.go
View file @ 0e1ee14c
......@@ -54,7 +54,8 @@ func NewResourceQuota(client client.Interface) admission.Interface {
return client.ResourceQuotas(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return client.ResourceQuotas(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), resourceVersion)
options := api.ListOptions{ResourceVersion: resourceVersion}
return client.ResourceQuotas(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
}
indexer, reflector := cache.NewNamespaceKeyedIndexerAndReflector(lw, &api.ResourceQuota{}, 0)
......
plugin/pkg/admission/serviceaccount/admission.go
View file @ 0e1ee14c
......@@ -86,7 +86,8 @@ func NewServiceAccount(cl client.Interface) *serviceAccount {
return cl.ServiceAccounts(api.NamespaceAll).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return cl.ServiceAccounts(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), resourceVersion)
options := api.ListOptions{ResourceVersion: resourceVersion}
return cl.ServiceAccounts(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.ServiceAccount{},
......@@ -100,7 +101,8 @@ func NewServiceAccount(cl client.Interface) *serviceAccount {
return cl.Secrets(api.NamespaceAll).List(labels.Everything(), tokenSelector)
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return cl.Secrets(api.NamespaceAll).Watch(labels.Everything(), tokenSelector, resourceVersion)
options := api.ListOptions{ResourceVersion: resourceVersion}
return cl.Secrets(api.NamespaceAll).Watch(labels.Everything(), tokenSelector, options)
},
},
&api.Secret{},
......
test/e2e/daemon_restart.go
View file @ 0e1ee14c
......@@ -225,7 +225,8 @@ var _ = Describe("DaemonRestart", func() {
return framework.Client.Pods(ns).List(labelSelector, fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return framework.Client.Pods(ns).Watch(labelSelector, fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return framework.Client.Pods(ns).Watch(labelSelector, fields.Everything(), options)
},
},
&api.Pod{},
......
test/e2e/density.go
View file @ 0e1ee14c
......@@ -203,7 +203,8 @@ var _ = Describe("Density", func() {
return c.Events(ns).List(labels.Everything(), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return c.Events(ns).Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return c.Events(ns).Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Event{},
......@@ -286,7 +287,8 @@ var _ = Describe("Density", func() {
return c.Pods(ns).List(labels.SelectorFromSet(labels.Set{"name": additionalPodsPrefix}), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return c.Pods(ns).Watch(labels.SelectorFromSet(labels.Set{"name": additionalPodsPrefix}), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return c.Pods(ns).Watch(labels.SelectorFromSet(labels.Set{"name": additionalPodsPrefix}), fields.Everything(), options)
},
},
&api.Pod{},
......
test/e2e/framework.go
View file @ 0e1ee14c
......@@ -153,7 +153,7 @@ func (f *Framework) WaitForAnEndpoint(serviceName string) error {
w, err := f.Client.Endpoints(f.Namespace.Name).Watch(
labels.Everything(),
fields.Set{"metadata.name": serviceName}.AsSelector(),
rv,
api.ListOptions{ResourceVersion: rv},
)
if err != nil {
return err
......
test/e2e/latency.go
View file @ 0e1ee14c
......@@ -153,7 +153,8 @@ func runLatencyTest(nodeCount int, c *client.Client, ns string) {
return c.Pods(ns).List(labels.SelectorFromSet(labels.Set{"name": additionalPodsPrefix}), fields.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return c.Pods(ns).Watch(labels.SelectorFromSet(labels.Set{"name": additionalPodsPrefix}), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return c.Pods(ns).Watch(labels.SelectorFromSet(labels.Set{"name": additionalPodsPrefix}), fields.Everything(), options)
},
},
&api.Pod{},
......
test/e2e/pods.go
View file @ 0e1ee14c
......@@ -304,7 +304,8 @@ var _ = Describe("Pods", func() {
}
Expect(len(pods.Items)).To(Equal(0))
w, err := podClient.Watch(
labels.SelectorFromSet(labels.Set(map[string]string{"time": value})), fields.Everything(), pods.ListMeta.ResourceVersion)
labels.SelectorFromSet(labels.Set(map[string]string{"time": value})), fields.Everything(),
api.ListOptions{ResourceVersion: pods.ListMeta.ResourceVersion})
if err != nil {
Failf("Failed to set up watch: %v", err)
}
......
test/e2e/service_latency.go
View file @ 0e1ee14c
......@@ -282,7 +282,8 @@ func startEndpointWatcher(f *Framework, q *endpointQueries) {
return f.Client.Endpoints(f.Namespace.Name).List(labels.Everything())
},
WatchFunc: func(rv string) (watch.Interface, error) {
return f.Client.Endpoints(f.Namespace.Name).Watch(labels.Everything(), fields.Everything(), rv)
options := api.ListOptions{ResourceVersion: rv}
return f.Client.Endpoints(f.Namespace.Name).Watch(labels.Everything(), fields.Everything(), options)
},
},
&api.Endpoints{},
......
test/e2e/util.go
View file @ 0e1ee14c
......@@ -145,7 +145,8 @@ func newPodStore(c *client.Client, namespace string, label labels.Selector, fiel
return c.Pods(namespace).List(label, field)
},
WatchFunc: func(rv string) (watch.Interface, error) {
return c.Pods(namespace).Watch(label, field, rv)
options := api.ListOptions{ResourceVersion: rv}
return c.Pods(namespace).Watch(label, field, options)
},
}
store := cache.NewStore(cache.MetaNamespaceKeyFunc)
......
test/integration/client_test.go
View file @ 0e1ee14c
......@@ -247,7 +247,7 @@ func TestMultiWatch(t *testing.T) {
w, err := client.Pods(ns).Watch(
labels.Set{"watchlabel": name}.AsSelector(),
fields.Everything(),
rv,
api.ListOptions{ResourceVersion: rv},
)
if err != nil {
panic(fmt.Sprintf("watch error for %v: %v", name, err))
......
test/integration/persistent_volumes_test.go
View file @ 0e1ee14c
......@@ -74,7 +74,7 @@ func TestPersistentVolumeRecycler(t *testing.T) {
},
}
w, _ := testClient.PersistentVolumes().Watch(labels.Everything(), fields.Everything(), "0")
w, _ := testClient.PersistentVolumes().Watch(labels.Everything(), fields.Everything(), api.ListOptions{})
defer w.Stop()
_, _ = testClient.PersistentVolumes().Create(pv)
......@@ -100,7 +100,7 @@ func TestPersistentVolumeRecycler(t *testing.T) {
// change the reclamation policy of the PV for the next test
pv.Spec.PersistentVolumeReclaimPolicy = api.PersistentVolumeReclaimDelete
w, _ = testClient.PersistentVolumes().Watch(labels.Everything(), fields.Everything(), "0")
w, _ = testClient.PersistentVolumes().Watch(labels.Everything(), fields.Everything(), api.ListOptions{})
defer w.Stop()
_, _ = testClient.PersistentVolumes().Create(pv)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment