diff --git a/cmd/e2e/e2e.go b/cmd/e2e/e2e.go index 89822fede5..d456f0a3e8 100644 --- a/cmd/e2e/e2e.go +++ b/cmd/e2e/e2e.go @@ -40,9 +40,8 @@ var ( func waitForPodRunning(c *client.Client, id string) { for { - ctx := api.NewContext() time.Sleep(5 * time.Second) - pod, err := c.GetPod(ctx, id) + pod, err := c.Pods(api.NamespaceDefault).Get(id) if err != nil { glog.Warningf("Get pod failed: %v", err) continue @@ -100,26 +99,26 @@ func loadClientOrDie() *client.Client { } func TestPodUpdate(c *client.Client) bool { - ctx := api.NewContext() + podClient := c.Pods(api.NamespaceDefault) pod := loadPodOrDie("./api/examples/pod.json") value := strconv.Itoa(time.Now().Nanosecond()) pod.Labels["time"] = value - _, err := c.CreatePod(ctx, pod) + _, err := podClient.Create(pod) if err != nil { glog.Errorf("Failed to create pod: %v", err) return false } - defer c.DeletePod(ctx, pod.Name) + defer podClient.Delete(pod.Name) waitForPodRunning(c, pod.Name) - pods, err := c.ListPods(ctx, labels.SelectorFromSet(labels.Set(map[string]string{"time": value}))) + pods, err := podClient.List(labels.SelectorFromSet(labels.Set(map[string]string{"time": value}))) if len(pods.Items) != 1 { glog.Errorf("Failed to find the correct pod") return false } - podOut, err := c.GetPod(ctx, pod.Name) + podOut, err := podClient.Get(pod.Name) if err != nil { glog.Errorf("Failed to get pod: %v", err) return false @@ -128,13 +127,13 @@ func TestPodUpdate(c *client.Client) bool { pod.Labels["time"] = value pod.ResourceVersion = podOut.ResourceVersion pod.DesiredState.Manifest.UUID = podOut.DesiredState.Manifest.UUID - pod, err = c.UpdatePod(ctx, pod) + pod, err = podClient.Update(pod) if err != nil { glog.Errorf("Failed to update pod: %v", err) return false } waitForPodRunning(c, pod.Name) - pods, err = c.ListPods(ctx, labels.SelectorFromSet(labels.Set(map[string]string{"time": value}))) + pods, err = podClient.List(labels.SelectorFromSet(labels.Set(map[string]string{"time": value}))) if len(pods.Items) != 1 { glog.Errorf("Failed to find the correct pod after update.") return false diff --git a/cmd/proxy/proxy.go b/cmd/proxy/proxy.go index 5731262e80..b9c3fdb1d9 100644 --- a/cmd/proxy/proxy.go +++ b/cmd/proxy/proxy.go @@ -21,6 +21,7 @@ import ( "net" "time" + "github.com/GoogleCloudPlatform/kubernetes/pkg/api" "github.com/GoogleCloudPlatform/kubernetes/pkg/client" "github.com/GoogleCloudPlatform/kubernetes/pkg/proxy" "github.com/GoogleCloudPlatform/kubernetes/pkg/proxy/config" @@ -63,7 +64,8 @@ func main() { glog.Fatalf("Invalid API configuration: %v", err) } config.NewSourceAPI( - client, + client.Services(api.NamespaceAll), + client.Endpoints(api.NamespaceAll), 30*time.Second, serviceConfig.Channel("api"), endpointsConfig.Channel("api"), diff --git a/pkg/client/client.go b/pkg/client/client.go index 2274f5df3b..b32a6bc832 100644 --- a/pkg/client/client.go +++ b/pkg/client/client.go @@ -21,64 +21,42 @@ import ( "fmt" "github.com/GoogleCloudPlatform/kubernetes/pkg/api" - "github.com/GoogleCloudPlatform/kubernetes/pkg/labels" "github.com/GoogleCloudPlatform/kubernetes/pkg/version" - "github.com/GoogleCloudPlatform/kubernetes/pkg/watch" ) // Interface holds the methods for clients of Kubernetes, // an interface to allow mock testing. -// TODO: these should return/take pointers. type Interface interface { - PodInterface - ReplicationControllerInterface - ServiceInterface + PodsNamespacer + ReplicationControllersNamespacer + ServicesNamespacer VersionInterface - MinionInterface + MinionsInterface + EventsInterface } -// PodInterface has methods to work with Pod resources. -type PodInterface interface { - ListPods(ctx api.Context, selector labels.Selector) (*api.PodList, error) - GetPod(ctx api.Context, id string) (*api.Pod, error) - DeletePod(ctx api.Context, id string) error - CreatePod(ctx api.Context, pod *api.Pod) (*api.Pod, error) - UpdatePod(ctx api.Context, pod *api.Pod) (*api.Pod, error) +func (c *Client) ReplicationControllers(namespace string) ReplicationControllerInterface { + return NewReplicationControllersClient(c, namespace) } -// ReplicationControllerInterface has methods to work with ReplicationController resources. -type ReplicationControllerInterface interface { - ListReplicationControllers(ctx api.Context, selector labels.Selector) (*api.ReplicationControllerList, error) - GetReplicationController(ctx api.Context, id string) (*api.ReplicationController, error) - CreateReplicationController(ctx api.Context, ctrl *api.ReplicationController) (*api.ReplicationController, error) - UpdateReplicationController(ctx api.Context, ctrl *api.ReplicationController) (*api.ReplicationController, error) - DeleteReplicationController(ctx api.Context, id string) error - WatchReplicationControllers(ctx api.Context, label, field labels.Selector, resourceVersion string) (watch.Interface, error) +func (c *Client) Minions() MinionInterface { + return NewMinionsClient(c) } -// ServiceInterface has methods to work with Service resources. -type ServiceInterface interface { - ListServices(ctx api.Context, selector labels.Selector) (*api.ServiceList, error) - GetService(ctx api.Context, id string) (*api.Service, error) - CreateService(ctx api.Context, srv *api.Service) (*api.Service, error) - UpdateService(ctx api.Context, srv *api.Service) (*api.Service, error) - DeleteService(ctx api.Context, id string) error - WatchServices(ctx api.Context, label, field labels.Selector, resourceVersion string) (watch.Interface, error) +func (c *Client) Events() EventInterface { + return NewEventsClient(c) } -// EndpointsInterface has methods to work with Endpoints resources -type EndpointsInterface interface { - ListEndpoints(ctx api.Context, selector labels.Selector) (*api.EndpointsList, error) - GetEndpoints(ctx api.Context, id string) (*api.Endpoints, error) - WatchEndpoints(ctx api.Context, label, field labels.Selector, resourceVersion string) (watch.Interface, error) +func (c *Client) Endpoints(namespace string) EndpointsInterface { + return NewEndpointsClient(c, namespace) } -// EventInterface has methods to work with Event resources -type EventInterface interface { - CreateEvent(event *api.Event) (*api.Event, error) - ListEvents(selector labels.Selector) (*api.EventList, error) - GetEvent(id string) (*api.Event, error) - WatchEvents(label, field labels.Selector, resourceVersion string) (watch.Interface, error) +func (c *Client) Pods(namespace string) PodInterface { + return NewPodsClient(c, namespace) +} + +func (c *Client) Services(namespace string) ServiceInterface { + return NewServicesClient(c, namespace) } // VersionInterface has a method to retrieve the server version. @@ -86,12 +64,6 @@ type VersionInterface interface { ServerVersion() (*version.Info, error) } -type MinionInterface interface { - CreateMinion(minion *api.Minion) (*api.Minion, error) - ListMinions() (*api.MinionList, error) - DeleteMinion(id string) error -} - // APIStatus is exposed by errors that can be converted to an api.Status object // for finer grained details. type APIStatus interface { @@ -103,190 +75,6 @@ type Client struct { *RESTClient } -// ListPods takes a selector, and returns the list of pods that match that selector. -func (c *Client) ListPods(ctx api.Context, selector labels.Selector) (result *api.PodList, err error) { - result = &api.PodList{} - err = c.Get().Namespace(api.Namespace(ctx)).Path("pods").SelectorParam("labels", selector).Do().Into(result) - return -} - -// GetPod takes the id of the pod, and returns the corresponding Pod object, and an error if it occurs -func (c *Client) GetPod(ctx api.Context, id string) (result *api.Pod, err error) { - result = &api.Pod{} - err = c.Get().Namespace(api.Namespace(ctx)).Path("pods").Path(id).Do().Into(result) - return -} - -// DeletePod takes the id of the pod, and returns an error if one occurs -func (c *Client) DeletePod(ctx api.Context, id string) error { - return c.Delete().Namespace(api.Namespace(ctx)).Path("pods").Path(id).Do().Error() -} - -// CreatePod takes the representation of a pod. Returns the server's representation of the pod, and an error, if it occurs. -func (c *Client) CreatePod(ctx api.Context, pod *api.Pod) (result *api.Pod, err error) { - result = &api.Pod{} - err = c.Post().Namespace(api.Namespace(ctx)).Path("pods").Body(pod).Do().Into(result) - return -} - -// UpdatePod takes the representation of a pod to update. Returns the server's representation of the pod, and an error, if it occurs. -func (c *Client) UpdatePod(ctx api.Context, pod *api.Pod) (result *api.Pod, err error) { - result = &api.Pod{} - if len(pod.ResourceVersion) == 0 { - err = fmt.Errorf("invalid update object, missing resource version: %v", pod) - return - } - err = c.Put().Namespace(api.Namespace(ctx)).Path("pods").Path(pod.Name).Body(pod).Do().Into(result) - return -} - -// ListReplicationControllers takes a selector, and returns the list of replication controllers that match that selector. -func (c *Client) ListReplicationControllers(ctx api.Context, selector labels.Selector) (result *api.ReplicationControllerList, err error) { - result = &api.ReplicationControllerList{} - err = c.Get().Namespace(api.Namespace(ctx)).Path("replicationControllers").SelectorParam("labels", selector).Do().Into(result) - return -} - -// GetReplicationController returns information about a particular replication controller. -func (c *Client) GetReplicationController(ctx api.Context, id string) (result *api.ReplicationController, err error) { - result = &api.ReplicationController{} - err = c.Get().Namespace(api.Namespace(ctx)).Path("replicationControllers").Path(id).Do().Into(result) - return -} - -// CreateReplicationController creates a new replication controller. -func (c *Client) CreateReplicationController(ctx api.Context, controller *api.ReplicationController) (result *api.ReplicationController, err error) { - result = &api.ReplicationController{} - err = c.Post().Namespace(api.Namespace(ctx)).Path("replicationControllers").Body(controller).Do().Into(result) - return -} - -// UpdateReplicationController updates an existing replication controller. -func (c *Client) UpdateReplicationController(ctx api.Context, controller *api.ReplicationController) (result *api.ReplicationController, err error) { - result = &api.ReplicationController{} - if len(controller.ResourceVersion) == 0 { - err = fmt.Errorf("invalid update object, missing resource version: %v", controller) - return - } - err = c.Put().Namespace(api.Namespace(ctx)).Path("replicationControllers").Path(controller.Name).Body(controller).Do().Into(result) - return -} - -// DeleteReplicationController deletes an existing replication controller. -func (c *Client) DeleteReplicationController(ctx api.Context, id string) error { - return c.Delete().Namespace(api.Namespace(ctx)).Path("replicationControllers").Path(id).Do().Error() -} - -// WatchReplicationControllers returns a watch.Interface that watches the requested controllers. -func (c *Client) WatchReplicationControllers(ctx api.Context, label, field labels.Selector, resourceVersion string) (watch.Interface, error) { - return c.Get(). - Namespace(api.Namespace(ctx)). - Path("watch"). - Path("replicationControllers"). - Param("resourceVersion", resourceVersion). - SelectorParam("labels", label). - SelectorParam("fields", field). - Watch() -} - -// ListServices takes a selector, and returns the list of services that match that selector -func (c *Client) ListServices(ctx api.Context, selector labels.Selector) (result *api.ServiceList, err error) { - result = &api.ServiceList{} - err = c.Get().Namespace(api.Namespace(ctx)).Path("services").SelectorParam("labels", selector).Do().Into(result) - return -} - -// GetService returns information about a particular service. -func (c *Client) GetService(ctx api.Context, id string) (result *api.Service, err error) { - result = &api.Service{} - err = c.Get().Namespace(api.Namespace(ctx)).Path("services").Path(id).Do().Into(result) - return -} - -// CreateService creates a new service. -func (c *Client) CreateService(ctx api.Context, svc *api.Service) (result *api.Service, err error) { - result = &api.Service{} - err = c.Post().Namespace(api.Namespace(ctx)).Path("services").Body(svc).Do().Into(result) - return -} - -// UpdateService updates an existing service. -func (c *Client) UpdateService(ctx api.Context, svc *api.Service) (result *api.Service, err error) { - result = &api.Service{} - if len(svc.ResourceVersion) == 0 { - err = fmt.Errorf("invalid update object, missing resource version: %v", svc) - return - } - err = c.Put().Namespace(api.Namespace(ctx)).Path("services").Path(svc.Name).Body(svc).Do().Into(result) - return -} - -// DeleteService deletes an existing service. -func (c *Client) DeleteService(ctx api.Context, id string) error { - return c.Delete().Namespace(api.Namespace(ctx)).Path("services").Path(id).Do().Error() -} - -// WatchServices returns a watch.Interface that watches the requested services. -func (c *Client) WatchServices(ctx api.Context, label, field labels.Selector, resourceVersion string) (watch.Interface, error) { - return c.Get(). - Namespace(api.Namespace(ctx)). - Path("watch"). - Path("services"). - Param("resourceVersion", resourceVersion). - SelectorParam("labels", label). - SelectorParam("fields", field). - Watch() -} - -// ListEndpoints takes a selector, and returns the list of endpoints that match that selector -func (c *Client) ListEndpoints(ctx api.Context, selector labels.Selector) (result *api.EndpointsList, err error) { - result = &api.EndpointsList{} - err = c.Get().Namespace(api.Namespace(ctx)).Path("endpoints").SelectorParam("labels", selector).Do().Into(result) - return -} - -// GetEndpoints returns information about the endpoints for a particular service. -func (c *Client) GetEndpoints(ctx api.Context, id string) (result *api.Endpoints, err error) { - result = &api.Endpoints{} - err = c.Get().Namespace(api.Namespace(ctx)).Path("endpoints").Path(id).Do().Into(result) - return -} - -// WatchEndpoints returns a watch.Interface that watches the requested endpoints for a service. -func (c *Client) WatchEndpoints(ctx api.Context, label, field labels.Selector, resourceVersion string) (watch.Interface, error) { - return c.Get(). - Namespace(api.Namespace(ctx)). - Path("watch"). - Path("endpoints"). - Param("resourceVersion", resourceVersion). - SelectorParam("labels", label). - SelectorParam("fields", field). - Watch() -} - -// CreateEndpoints creates a new endpoint. -func (c *Client) CreateEndpoints(ctx api.Context, endpoints *api.Endpoints) (*api.Endpoints, error) { - result := &api.Endpoints{} - err := c.Post().Namespace(api.Namespace(ctx)).Path("endpoints").Body(endpoints).Do().Into(result) - return result, err -} - -// UpdateEndpoints updates an existing endpoint. -func (c *Client) UpdateEndpoints(ctx api.Context, endpoints *api.Endpoints) (*api.Endpoints, error) { - result := &api.Endpoints{} - if len(endpoints.ResourceVersion) == 0 { - return nil, fmt.Errorf("invalid update object, missing resource version: %v", endpoints) - } - err := c.Put(). - Namespace(api.Namespace(ctx)). - Path("endpoints"). - Path(endpoints.Name). - Body(endpoints). - Do(). - Into(result) - return result, err -} - // ServerVersion retrieves and parses the server's version. func (c *Client) ServerVersion() (*version.Info, error) { body, err := c.Get().AbsPath("/version").Do().Raw() @@ -300,66 +88,3 @@ func (c *Client) ServerVersion() (*version.Info, error) { } return &info, nil } - -// CreateMinion creates a new minion. -func (c *Client) CreateMinion(minion *api.Minion) (*api.Minion, error) { - result := &api.Minion{} - err := c.Post().Path("minions").Body(minion).Do().Into(result) - return result, err -} - -// ListMinions lists all the minions in the cluster. -func (c *Client) ListMinions() (*api.MinionList, error) { - result := &api.MinionList{} - err := c.Get().Path("minions").Do().Into(result) - return result, err -} - -// GetMinion returns information about a particular minion. -func (c *Client) GetMinion(id string) (*api.Minion, error) { - result := &api.Minion{} - err := c.Get().Path("minions").Path(id).Do().Into(result) - return result, err -} - -// DeleteMinion deletes an existing minion. -func (c *Client) DeleteMinion(id string) error { - return c.Delete().Path("minions").Path(id).Do().Error() -} - -// CreateEvent makes a new event. Returns the copy of the event the server returns, or an error. -func (c *Client) CreateEvent(event *api.Event) (*api.Event, error) { - result := &api.Event{} - err := c.Post().Path("events").Body(event).Do().Into(result) - return result, err -} - -// ListEvents returns a list of events matching the selectors. -func (c *Client) ListEvents(label, field labels.Selector) (*api.EventList, error) { - result := &api.EventList{} - err := c.Get(). - Path("events"). - SelectorParam("labels", label). - SelectorParam("fields", field). - Do(). - Into(result) - return result, err -} - -// GetEvent returns the given event, or an error. -func (c *Client) GetEvent(id string) (*api.Event, error) { - result := &api.Event{} - err := c.Get().Path("events").Path(id).Do().Into(result) - return result, err -} - -// WatchEvents starts watching for events matching the given selectors. -func (c *Client) WatchEvents(label, field labels.Selector, resourceVersion string) (watch.Interface, error) { - return c.Get(). - Path("watch"). - Path("events"). - Param("resourceVersion", resourceVersion). - SelectorParam("labels", label). - SelectorParam("fields", field). - Watch() -} diff --git a/pkg/client/conditions.go b/pkg/client/conditions.go index 877263dedf..743bea902a 100644 --- a/pkg/client/conditions.go +++ b/pkg/client/conditions.go @@ -26,8 +26,7 @@ import ( // for a controller's ReplicaSelector equals the Replicas count. func (c *Client) ControllerHasDesiredReplicas(controller api.ReplicationController) wait.ConditionFunc { return func() (bool, error) { - ctx := api.WithNamespace(api.NewContext(), controller.Namespace) - pods, err := c.ListPods(ctx, labels.Set(controller.DesiredState.ReplicaSelector).AsSelector()) + pods, err := c.Pods(controller.Namespace).List(labels.Set(controller.DesiredState.ReplicaSelector).AsSelector()) if err != nil { return false, err } diff --git a/pkg/client/endpoints.go b/pkg/client/endpoints.go new file mode 100644 index 0000000000..3ed21c1f9a --- /dev/null +++ b/pkg/client/endpoints.go @@ -0,0 +1,98 @@ +/* +Copyright 2014 Google Inc. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package client + +import ( + "fmt" + + "github.com/GoogleCloudPlatform/kubernetes/pkg/api" + "github.com/GoogleCloudPlatform/kubernetes/pkg/labels" + "github.com/GoogleCloudPlatform/kubernetes/pkg/watch" +) + +// EndpointsNamespacer has methods to work with Endpoint resources in a namespace +type EndpointsNamespacer interface { + Endpoint(namespace string) EndpointsInterface +} + +// EndpointsInterface has methods to work with Endpoints resources +type EndpointsInterface interface { + Create(endpoints *api.Endpoints) (*api.Endpoints, error) + List(selector labels.Selector) (*api.EndpointsList, error) + Get(id string) (*api.Endpoints, error) + Update(endpoints *api.Endpoints) (*api.Endpoints, error) + Watch(label, field labels.Selector, resourceVersion string) (watch.Interface, error) +} + +// EndpointsClient implements EndpointsNamespacer interface +type EndpointsClient struct { + r *Client + ns string +} + +// NewEndpointsClient returns a EndpointsClient +func NewEndpointsClient(c *Client, namespace string) *EndpointsClient { + return &EndpointsClient{c, namespace} +} + +// Create creates a new endpoint. +func (c *EndpointsClient) Create(endpoints *api.Endpoints) (*api.Endpoints, error) { + result := &api.Endpoints{} + err := c.r.Post().Namespace(c.ns).Path("endpoints").Body(endpoints).Do().Into(result) + return result, err +} + +// List takes a selector, and returns the list of endpoints that match that selector +func (c *EndpointsClient) List(selector labels.Selector) (result *api.EndpointsList, err error) { + result = &api.EndpointsList{} + err = c.r.Get().Namespace(c.ns).Path("endpoints").SelectorParam("labels", selector).Do().Into(result) + return +} + +// Get returns information about the endpoints for a particular service. +func (c *EndpointsClient) Get(id string) (result *api.Endpoints, err error) { + result = &api.Endpoints{} + err = c.r.Get().Namespace(c.ns).Path("endpoints").Path(id).Do().Into(result) + return +} + +// Watch returns a watch.Interface that watches the requested endpoints for a service. +func (c *EndpointsClient) Watch(label, field labels.Selector, resourceVersion string) (watch.Interface, error) { + return c.r.Get(). + Namespace(c.ns). + Path("watch"). + Path("endpoints"). + Param("resourceVersion", resourceVersion). + SelectorParam("labels", label). + SelectorParam("fields", field). + Watch() +} + +func (c *EndpointsClient) Update(endpoints *api.Endpoints) (*api.Endpoints, error) { + result := &api.Endpoints{} + if len(endpoints.ResourceVersion) == 0 { + return nil, fmt.Errorf("invalid update object, missing resource version: %v", endpoints) + } + err := c.r.Put(). + Namespace(c.ns). + Path("endpoints"). + Path(endpoints.Name). + Body(endpoints). + Do(). + Into(result) + return result, err +} diff --git a/pkg/client/events.go b/pkg/client/events.go new file mode 100644 index 0000000000..81eff7368f --- /dev/null +++ b/pkg/client/events.go @@ -0,0 +1,85 @@ +/* +Copyright 2014 Google Inc. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package client + +import ( + "github.com/GoogleCloudPlatform/kubernetes/pkg/api" + "github.com/GoogleCloudPlatform/kubernetes/pkg/labels" + "github.com/GoogleCloudPlatform/kubernetes/pkg/watch" +) + +// Events has methods to work with Event resources +type EventsInterface interface { + Events() EventInterface +} + +// EventInterface has methods to work with Event resources +type EventInterface interface { + Create(event *api.Event) (*api.Event, error) + List(label, field labels.Selector) (*api.EventList, error) + Get(id string) (*api.Event, error) + Watch(label, field labels.Selector, resourceVersion string) (watch.Interface, error) +} + +// EventsClient implements Events interface +type EventsClient struct { + r *Client +} + +// NewEventsClient returns a EventsClient +func NewEventsClient(c *Client) *EventsClient { + return &EventsClient{ + r: c, + } +} + +// Create makes a new event. Returns the copy of the event the server returns, or an error. +func (c *EventsClient) Create(event *api.Event) (*api.Event, error) { + result := &api.Event{} + err := c.r.Post().Path("events").Body(event).Do().Into(result) + return result, err +} + +// List returns a list of events matching the selectors. +func (c *EventsClient) List(label, field labels.Selector) (*api.EventList, error) { + result := &api.EventList{} + err := c.r.Get(). + Path("events"). + SelectorParam("labels", label). + SelectorParam("fields", field). + Do(). + Into(result) + return result, err +} + +// Get returns the given event, or an error. +func (c *EventsClient) Get(id string) (*api.Event, error) { + result := &api.Event{} + err := c.r.Get().Path("events").Path(id).Do().Into(result) + return result, err +} + +// Watch starts watching for events matching the given selectors. +func (c *EventsClient) Watch(label, field labels.Selector, resourceVersion string) (watch.Interface, error) { + return c.r.Get(). + Path("watch"). + Path("events"). + Param("resourceVersion", resourceVersion). + SelectorParam("labels", label). + SelectorParam("fields", field). + Watch() +} diff --git a/pkg/client/minions.go b/pkg/client/minions.go new file mode 100644 index 0000000000..23e7015d9c --- /dev/null +++ b/pkg/client/minions.go @@ -0,0 +1,68 @@ +/* +Copyright 2014 Google Inc. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package client + +import ( + "github.com/GoogleCloudPlatform/kubernetes/pkg/api" +) + +type MinionsInterface interface { + Minions() MinionInterface +} + +type MinionInterface interface { + Get(id string) (result *api.Minion, err error) + Create(minion *api.Minion) (*api.Minion, error) + List() (*api.MinionList, error) + Delete(id string) error +} + +// MinionsClient implements Minions interface +type MinionsClient struct { + r *Client +} + +// NewMinionsClient returns a MinionsClient +func NewMinionsClient(c *Client) *MinionsClient { + return &MinionsClient{c} +} + +// Create creates a new minion. +func (c *MinionsClient) Create(minion *api.Minion) (*api.Minion, error) { + result := &api.Minion{} + err := c.r.Post().Path("minions").Body(minion).Do().Into(result) + return result, err +} + +// List lists all the minions in the cluster. +func (c *MinionsClient) List() (result *api.MinionList, err error) { + result = &api.MinionList{} + err = c.r.Get().Path("minions").Do().Into(result) + return +} + +// Get gets an existing minion +func (c *MinionsClient) Get(id string) (result *api.Minion, err error) { + result = &api.Minion{} + err = c.r.Get().Path("minions").Path(id).Do().Into(result) + return +} + +// Delete deletes an existing minion. +func (c *MinionsClient) Delete(id string) error { + return c.r.Delete().Path("minions").Path(id).Do().Error() +} diff --git a/pkg/client/pods.go b/pkg/client/pods.go new file mode 100644 index 0000000000..c9e5ab34fc --- /dev/null +++ b/pkg/client/pods.go @@ -0,0 +1,89 @@ +/* +Copyright 2014 Google Inc. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package client + +import ( + "fmt" + + "github.com/GoogleCloudPlatform/kubernetes/pkg/api" + "github.com/GoogleCloudPlatform/kubernetes/pkg/labels" +) + +// PodsNamespacer has methods to work with Pod resources in a namespace +type PodsNamespacer interface { + Pods(namespace string) PodInterface +} + +// PodInterface has methods to work with Pod resources. +type PodInterface interface { + List(selector labels.Selector) (*api.PodList, error) + Get(name string) (*api.Pod, error) + Delete(name string) error + Create(pod *api.Pod) (*api.Pod, error) + Update(pod *api.Pod) (*api.Pod, error) +} + +// PodsClient implements PodsNamespacer interface +type PodsClient struct { + r *Client + ns string +} + +// NewPodsClient returns a PodsClient +func NewPodsClient(c *Client, namespace string) *PodsClient { + return &PodsClient{ + r: c, + ns: namespace, + } +} + +// ListPods takes a selector, and returns the list of pods that match that selector. +func (c *PodsClient) List(selector labels.Selector) (result *api.PodList, err error) { + result = &api.PodList{} + err = c.r.Get().Namespace(c.ns).Path("pods").SelectorParam("labels", selector).Do().Into(result) + return +} + +// GetPod takes the name of the pod, and returns the corresponding Pod object, and an error if it occurs +func (c *PodsClient) Get(name string) (result *api.Pod, err error) { + result = &api.Pod{} + err = c.r.Get().Namespace(c.ns).Path("pods").Path(name).Do().Into(result) + return +} + +// DeletePod takes the name of the pod, and returns an error if one occurs +func (c *PodsClient) Delete(name string) error { + return c.r.Delete().Namespace(c.ns).Path("pods").Path(name).Do().Error() +} + +// CreatePod takes the representation of a pod. Returns the server's representation of the pod, and an error, if it occurs. +func (c *PodsClient) Create(pod *api.Pod) (result *api.Pod, err error) { + result = &api.Pod{} + err = c.r.Post().Namespace(c.ns).Path("pods").Body(pod).Do().Into(result) + return +} + +// UpdatePod takes the representation of a pod to update. Returns the server's representation of the pod, and an error, if it occurs. +func (c *PodsClient) Update(pod *api.Pod) (result *api.Pod, err error) { + result = &api.Pod{} + if len(pod.ResourceVersion) == 0 { + err = fmt.Errorf("invalid update object, missing resource version: %v", pod) + return + } + err = c.r.Put().Namespace(c.ns).Path("pods").Path(pod.Name).Body(pod).Do().Into(result) + return +} diff --git a/pkg/client/record/event.go b/pkg/client/record/event.go index eec64a0936..6d7f994b8b 100644 --- a/pkg/client/record/event.go +++ b/pkg/client/record/event.go @@ -30,7 +30,7 @@ import ( // EventRecorder knows how to store events (client.Client implements it.) type EventRecorder interface { - CreateEvent(event *api.Event) (*api.Event, error) + Create(event *api.Event) (*api.Event, error) } // StartRecording starts sending events to recorder. Call once while initializing @@ -44,7 +44,7 @@ func StartRecording(recorder EventRecorder, sourceName string) watch.Interface { event = &eventCopy event.Source = sourceName for { - _, err := recorder.CreateEvent(event) + _, err := recorder.Create(event) if err == nil { break } diff --git a/pkg/client/replication_controllers.go b/pkg/client/replication_controllers.go new file mode 100644 index 0000000000..8c7efb1bdf --- /dev/null +++ b/pkg/client/replication_controllers.go @@ -0,0 +1,100 @@ +/* +Copyright 2014 Google Inc. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package client + +import ( + "fmt" + + "github.com/GoogleCloudPlatform/kubernetes/pkg/api" + "github.com/GoogleCloudPlatform/kubernetes/pkg/labels" + "github.com/GoogleCloudPlatform/kubernetes/pkg/watch" +) + +// ReplicationControllersNamespacer has methods to work with ReplicationController resources in a namespace +type ReplicationControllersNamespacer interface { + ReplicationControllers(namespace string) ReplicationControllerInterface +} + +// ReplicationControllerInterface has methods to work with ReplicationController resources. +type ReplicationControllerInterface interface { + List(selector labels.Selector) (*api.ReplicationControllerList, error) + Get(name string) (*api.ReplicationController, error) + Create(ctrl *api.ReplicationController) (*api.ReplicationController, error) + Update(ctrl *api.ReplicationController) (*api.ReplicationController, error) + Delete(name string) error + Watch(label, field labels.Selector, resourceVersion string) (watch.Interface, error) +} + +// ReplicationControllersClient implements ReplicationControllersNamespacer interface +type ReplicationControllersClient struct { + r *Client + ns string +} + +// NewReplicationControllersClient returns a PodsClient +func NewReplicationControllersClient(c *Client, namespace string) *ReplicationControllersClient { + return &ReplicationControllersClient{c, namespace} +} + +// List takes a selector, and returns the list of replication controllers that match that selector. +func (c *ReplicationControllersClient) List(selector labels.Selector) (result *api.ReplicationControllerList, err error) { + result = &api.ReplicationControllerList{} + err = c.r.Get().Namespace(c.ns).Path("replicationControllers").SelectorParam("labels", selector).Do().Into(result) + return +} + +// Get returns information about a particular replication controller. +func (c *ReplicationControllersClient) Get(name string) (result *api.ReplicationController, err error) { + result = &api.ReplicationController{} + err = c.r.Get().Namespace(c.ns).Path("replicationControllers").Path(name).Do().Into(result) + return +} + +// Create creates a new replication controller. +func (c *ReplicationControllersClient) Create(controller *api.ReplicationController) (result *api.ReplicationController, err error) { + result = &api.ReplicationController{} + err = c.r.Post().Namespace(c.ns).Path("replicationControllers").Body(controller).Do().Into(result) + return +} + +// Update updates an existing replication controller. +func (c *ReplicationControllersClient) Update(controller *api.ReplicationController) (result *api.ReplicationController, err error) { + result = &api.ReplicationController{} + if len(controller.ResourceVersion) == 0 { + err = fmt.Errorf("invalid update object, missing resource version: %v", controller) + return + } + err = c.r.Put().Namespace(c.ns).Path("replicationControllers").Path(controller.Name).Body(controller).Do().Into(result) + return +} + +// Delete deletes an existing replication controller. +func (c *ReplicationControllersClient) Delete(name string) error { + return c.r.Delete().Namespace(c.ns).Path("replicationControllers").Path(name).Do().Error() +} + +// Watch returns a watch.Interface that watches the requested controllers. +func (c *ReplicationControllersClient) Watch(label, field labels.Selector, resourceVersion string) (watch.Interface, error) { + return c.r.Get(). + Namespace(c.ns). + Path("watch"). + Path("replicationControllers"). + Param("resourceVersion", resourceVersion). + SelectorParam("labels", label). + SelectorParam("fields", field). + Watch() +} diff --git a/pkg/client/services.go b/pkg/client/services.go new file mode 100644 index 0000000000..872fd53512 --- /dev/null +++ b/pkg/client/services.go @@ -0,0 +1,100 @@ +/* +Copyright 2014 Google Inc. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package client + +import ( + "fmt" + + "github.com/GoogleCloudPlatform/kubernetes/pkg/api" + "github.com/GoogleCloudPlatform/kubernetes/pkg/labels" + "github.com/GoogleCloudPlatform/kubernetes/pkg/watch" +) + +// ServicesNamespacer has methods to work with Service resources in a namespace +type ServicesNamespacer interface { + Services(namespace string) ServiceInterface +} + +// ServiceInterface has methods to work with Service resources. +type ServiceInterface interface { + List(selector labels.Selector) (*api.ServiceList, error) + Get(name string) (*api.Service, error) + Create(srv *api.Service) (*api.Service, error) + Update(srv *api.Service) (*api.Service, error) + Delete(name string) error + Watch(label, field labels.Selector, resourceVersion string) (watch.Interface, error) +} + +// ServicesClient implements PodsNamespacer interface +type ServicesClient struct { + r *Client + ns string +} + +// NewServicesClient returns a PodsClient +func NewServicesClient(c *Client, namespace string) *ServicesClient { + return &ServicesClient{c, namespace} +} + +// List takes a selector, and returns the list of services that match that selector +func (c *ServicesClient) List(selector labels.Selector) (result *api.ServiceList, err error) { + result = &api.ServiceList{} + err = c.r.Get().Namespace(c.ns).Path("services").SelectorParam("labels", selector).Do().Into(result) + return +} + +// Get returns information about a particular service. +func (c *ServicesClient) Get(name string) (result *api.Service, err error) { + result = &api.Service{} + err = c.r.Get().Namespace(c.ns).Path("services").Path(name).Do().Into(result) + return +} + +// Create creates a new service. +func (c *ServicesClient) Create(svc *api.Service) (result *api.Service, err error) { + result = &api.Service{} + err = c.r.Post().Namespace(c.ns).Path("services").Body(svc).Do().Into(result) + return +} + +// Update updates an existing service. +func (c *ServicesClient) Update(svc *api.Service) (result *api.Service, err error) { + result = &api.Service{} + if len(svc.ResourceVersion) == 0 { + err = fmt.Errorf("invalid update object, missing resource version: %v", svc) + return + } + err = c.r.Put().Namespace(c.ns).Path("services").Path(svc.Name).Body(svc).Do().Into(result) + return +} + +// Delete deletes an existing service. +func (c *ServicesClient) Delete(name string) error { + return c.r.Delete().Namespace(c.ns).Path("services").Path(name).Do().Error() +} + +// Watch returns a watch.Interface that watches the requested services. +func (c *ServicesClient) Watch(label, field labels.Selector, resourceVersion string) (watch.Interface, error) { + return c.r.Get(). + Namespace(c.ns). + Path("watch"). + Path("services"). + Param("resourceVersion", resourceVersion). + SelectorParam("labels", label). + SelectorParam("fields", field). + Watch() +} diff --git a/pkg/controller/replication_controller.go b/pkg/controller/replication_controller.go index 1e67a8d640..db57ab5302 100644 --- a/pkg/controller/replication_controller.go +++ b/pkg/controller/replication_controller.go @@ -42,9 +42,9 @@ type ReplicationManager struct { // created as an interface to allow testing. type PodControlInterface interface { // createReplica creates new replicated pods according to the spec. - createReplica(ctx api.Context, controllerSpec api.ReplicationController) + createReplica(namespace string, controllerSpec api.ReplicationController) // deletePod deletes the pod identified by podID. - deletePod(ctx api.Context, podID string) error + deletePod(namespace string, podID string) error } // RealPodControl is the default implementation of PodControllerInterface. @@ -52,7 +52,7 @@ type RealPodControl struct { kubeClient client.Interface } -func (r RealPodControl) createReplica(ctx api.Context, controllerSpec api.ReplicationController) { +func (r RealPodControl) createReplica(namespace string, controllerSpec api.ReplicationController) { desiredLabels := make(labels.Set) for k, v := range controllerSpec.DesiredState.PodTemplate.Labels { desiredLabels[k] = v @@ -65,13 +65,13 @@ func (r RealPodControl) createReplica(ctx api.Context, controllerSpec api.Replic }, DesiredState: controllerSpec.DesiredState.PodTemplate.DesiredState, } - if _, err := r.kubeClient.CreatePod(ctx, pod); err != nil { + if _, err := r.kubeClient.Pods(namespace).Create(pod); err != nil { glog.Errorf("Unable to create pod replica: %v", err) } } -func (r RealPodControl) deletePod(ctx api.Context, podID string) error { - return r.kubeClient.DeletePod(ctx, podID) +func (r RealPodControl) deletePod(namespace, podID string) error { + return r.kubeClient.Pods(namespace).Delete(podID) } // NewReplicationManager creates a new ReplicationManager. @@ -95,9 +95,7 @@ func (rm *ReplicationManager) Run(period time.Duration) { // resourceVersion is a pointer to the resource version to use/update. func (rm *ReplicationManager) watchControllers(resourceVersion *string) { - ctx := api.NewContext() - watching, err := rm.kubeClient.WatchReplicationControllers( - ctx, + watching, err := rm.kubeClient.ReplicationControllers(api.NamespaceAll).Watch( labels.Everything(), labels.Everything(), *resourceVersion, @@ -147,8 +145,7 @@ func (rm *ReplicationManager) filterActivePods(pods []api.Pod) []api.Pod { func (rm *ReplicationManager) syncReplicationController(controllerSpec api.ReplicationController) error { s := labels.Set(controllerSpec.DesiredState.ReplicaSelector).AsSelector() - ctx := api.WithNamespace(api.NewContext(), controllerSpec.Namespace) - podList, err := rm.kubeClient.ListPods(ctx, s) + podList, err := rm.kubeClient.Pods(controllerSpec.Namespace).List(s) if err != nil { return err } @@ -162,7 +159,7 @@ func (rm *ReplicationManager) syncReplicationController(controllerSpec api.Repli for i := 0; i < diff; i++ { go func() { defer wait.Done() - rm.podControl.createReplica(ctx, controllerSpec) + rm.podControl.createReplica(controllerSpec.Namespace, controllerSpec) }() } wait.Wait() @@ -173,7 +170,7 @@ func (rm *ReplicationManager) syncReplicationController(controllerSpec api.Repli for i := 0; i < diff; i++ { go func(ix int) { defer wait.Done() - rm.podControl.deletePod(ctx, filteredList[ix].Name) + rm.podControl.deletePod(controllerSpec.Namespace, filteredList[ix].Name) }(i) } wait.Wait() @@ -185,8 +182,7 @@ func (rm *ReplicationManager) synchronize() { // TODO: remove this method completely and rely on the watch. // Add resource version tracking to watch to make this work. var controllerSpecs []api.ReplicationController - ctx := api.NewContext() - list, err := rm.kubeClient.ListReplicationControllers(ctx, labels.Everything()) + list, err := rm.kubeClient.ReplicationControllers(api.NamespaceAll).List(labels.Everything()) if err != nil { glog.Errorf("Synchronization error: %v (%#v)", err, err) return diff --git a/pkg/kubecfg/kubecfg.go b/pkg/kubecfg/kubecfg.go index 885f7756c4..d5e9aef794 100644 --- a/pkg/kubecfg/kubecfg.go +++ b/pkg/kubecfg/kubecfg.go @@ -127,14 +127,15 @@ func SaveNamespaceInfo(path string, ns *NamespaceInfo) error { // updating more complex replication controllers. If this is blank then no // update of the image is performed. func Update(ctx api.Context, name string, client client.Interface, updatePeriod time.Duration, imageName string) error { - controller, err := client.GetReplicationController(ctx, name) + // TODO ctx is not needed as input to this function, should just be 'namespace' + controller, err := client.ReplicationControllers(api.Namespace(ctx)).Get(name) if err != nil { return err } if len(imageName) != 0 { controller.DesiredState.PodTemplate.DesiredState.Manifest.Containers[0].Image = imageName - controller, err = client.UpdateReplicationController(ctx, controller) + controller, err = client.ReplicationControllers(controller.Namespace).Update(controller) if err != nil { return err } @@ -142,7 +143,7 @@ func Update(ctx api.Context, name string, client client.Interface, updatePeriod s := labels.Set(controller.DesiredState.ReplicaSelector).AsSelector() - podList, err := client.ListPods(ctx, s) + podList, err := client.Pods(api.Namespace(ctx)).List(s) if err != nil { return err } @@ -153,14 +154,14 @@ func Update(ctx api.Context, name string, client client.Interface, updatePeriod for _, pod := range podList.Items { // We delete the pod here, the controller will recreate it. This will result in pulling // a new Docker image. This isn't a full "update" but it's what we support for now. - err = client.DeletePod(ctx, pod.Name) + err = client.Pods(pod.Namespace).Delete(pod.Name) if err != nil { return err } time.Sleep(updatePeriod) } return wait.Poll(time.Second*5, time.Second*300, func() (bool, error) { - podList, err := client.ListPods(ctx, s) + podList, err := client.Pods(api.Namespace(ctx)).List(s) if err != nil { return false, err } @@ -175,12 +176,13 @@ func StopController(ctx api.Context, name string, client client.Interface) error // ResizeController resizes a controller named 'name' by setting replicas to 'replicas'. func ResizeController(ctx api.Context, name string, replicas int, client client.Interface) error { - controller, err := client.GetReplicationController(ctx, name) + // TODO ctx is not needed, and should just be a namespace + controller, err := client.ReplicationControllers(api.Namespace(ctx)).Get(name) if err != nil { return err } controller.DesiredState.Replicas = replicas - controllerOut, err := client.UpdateReplicationController(ctx, controller) + controllerOut, err := client.ReplicationControllers(api.Namespace(ctx)).Update(controller) if err != nil { return err } @@ -237,6 +239,7 @@ func portsFromString(spec string) ([]api.Port, error) { // RunController creates a new replication controller named 'name' which creates 'replicas' pods running 'image'. func RunController(ctx api.Context, image, name string, replicas int, client client.Interface, portSpec string, servicePort int) error { + // TODO replace ctx with a namespace string if servicePort > 0 && !util.IsDNSLabel(name) { return fmt.Errorf("Service creation requested, but an invalid name for a service was provided (%s). Service names must be valid DNS labels.", name) } @@ -273,7 +276,7 @@ func RunController(ctx api.Context, image, name string, replicas int, client cli }, } - controllerOut, err := client.CreateReplicationController(ctx, controller) + controllerOut, err := client.ReplicationControllers(api.Namespace(ctx)).Create(controller) if err != nil { return err } @@ -298,6 +301,7 @@ func RunController(ctx api.Context, image, name string, replicas int, client cli } func createService(ctx api.Context, name string, port int, client client.Interface) (*api.Service, error) { + // TODO remove context in favor of just namespace string svc := &api.Service{ ObjectMeta: api.ObjectMeta{ Name: name, @@ -310,19 +314,20 @@ func createService(ctx api.Context, name string, port int, client client.Interfa "simpleService": name, }, } - svc, err := client.CreateService(ctx, svc) + svc, err := client.Services(api.Namespace(ctx)).Create(svc) return svc, err } // DeleteController deletes a replication controller named 'name', requires that the controller // already be stopped. func DeleteController(ctx api.Context, name string, client client.Interface) error { - controller, err := client.GetReplicationController(ctx, name) + // TODO remove ctx in favor of just namespace string + controller, err := client.ReplicationControllers(api.Namespace(ctx)).Get(name) if err != nil { return err } if controller.DesiredState.Replicas != 0 { return fmt.Errorf("controller has non-zero replicas (%d), please stop it first", controller.DesiredState.Replicas) } - return client.DeleteReplicationController(ctx, name) + return client.ReplicationControllers(api.Namespace(ctx)).Delete(name) } diff --git a/pkg/kubectl/describe.go b/pkg/kubectl/describe.go index ed3990a1bd..6697c7b152 100644 --- a/pkg/kubectl/describe.go +++ b/pkg/kubectl/describe.go @@ -55,7 +55,8 @@ func Describe(w io.Writer, c client.Interface, resource, id string) error { } func describePod(w io.Writer, c client.Interface, id string) (string, error) { - pod, err := c.GetPod(api.NewDefaultContext(), id) + // TODO this needs proper namespace support + pod, err := c.Pods(api.NamespaceDefault).Get(id) if err != nil { return "", err } @@ -72,7 +73,8 @@ func describePod(w io.Writer, c client.Interface, id string) (string, error) { } func describeReplicationController(w io.Writer, c client.Interface, id string) (string, error) { - controller, err := c.GetReplicationController(api.NewDefaultContext(), id) + // TODO this needs proper namespace support + controller, err := c.ReplicationControllers(api.NamespaceDefault).Get(id) if err != nil { return "", err } @@ -94,7 +96,7 @@ func describeReplicationController(w io.Writer, c client.Interface, id string) ( } func describeService(w io.Writer, c client.Interface, id string) (string, error) { - service, err := c.GetService(api.NewDefaultContext(), id) + service, err := c.Services(api.NamespaceDefault).Get(id) if err != nil { return "", err } @@ -122,7 +124,7 @@ func describeMinion(w io.Writer, c client.Interface, id string) (string, error) // client.Interface doesn't have GetMinion(id) yet so we hack it up. func getMinion(c client.Interface, id string) (*api.Minion, error) { - minionList, err := c.ListMinions() + minionList, err := c.Minions().List() if err != nil { glog.Fatalf("Error getting minion info: %v\n", err) } @@ -141,7 +143,8 @@ func getMinion(c client.Interface, id string) (*api.Minion, error) { // of getting all RC's and searching through them manually). func getReplicationControllersForLabels(c client.Interface, labelsToMatch labels.Labels) string { // Get all replication controllers. - rcs, err := c.ListReplicationControllers(api.NewDefaultContext(), labels.Everything()) + // TODO this needs a namespace scope as argument + rcs, err := c.ReplicationControllers(api.NamespaceDefault).List(labels.Everything()) if err != nil { glog.Fatalf("Error getting replication controllers: %v\n", err) } @@ -169,7 +172,7 @@ func getReplicationControllersForLabels(c client.Interface, labelsToMatch labels } func getPodStatusForReplicationController(kubeClient client.Interface, controller *api.ReplicationController) (running, waiting, terminated int, err error) { - rcPods, err := kubeClient.ListPods(api.NewDefaultContext(), labels.SelectorFromSet(controller.DesiredState.ReplicaSelector)) + rcPods, err := kubeClient.Pods(controller.Namespace).List(labels.SelectorFromSet(controller.DesiredState.ReplicaSelector)) if err != nil { return } diff --git a/pkg/master/master.go b/pkg/master/master.go index bc708fdbc1..0db5ca52ba 100644 --- a/pkg/master/master.go +++ b/pkg/master/master.go @@ -137,7 +137,7 @@ func (m *Master) init(c *Config) { PodCache: podCache, PodInfoGetter: c.KubeletClient, Registry: m.podRegistry, - Minions: m.client, + Minions: m.client.Minions(), }), "replicationControllers": controller.NewREST(m.controllerRegistry, m.podRegistry), "services": service.NewREST(m.serviceRegistry, c.Cloud, m.minionRegistry, m.portalNet), diff --git a/pkg/proxy/config/api.go b/pkg/proxy/config/api.go index e62c122e3f..77a54ab27e 100644 --- a/pkg/proxy/config/api.go +++ b/pkg/proxy/config/api.go @@ -27,18 +27,24 @@ import ( "github.com/golang/glog" ) -// Watcher is the interface needed to receive changes to services and endpoints. -type Watcher interface { - ListServices(ctx api.Context, label labels.Selector) (*api.ServiceList, error) - ListEndpoints(ctx api.Context, label labels.Selector) (*api.EndpointsList, error) - WatchServices(ctx api.Context, label, field labels.Selector, resourceVersion string) (watch.Interface, error) - WatchEndpoints(ctx api.Context, label, field labels.Selector, resourceVersion string) (watch.Interface, error) +// ServicesWatcher is capable of listing and watching for changes to services across ALL namespaces +type ServicesWatcher interface { + List(label labels.Selector) (*api.ServiceList, error) + Watch(label, field labels.Selector, resourceVersion string) (watch.Interface, error) +} + +// EndpointsWatcher is capable of listing and watching for changes to endpoints across ALL namespaces +type EndpointsWatcher interface { + List(label labels.Selector) (*api.EndpointsList, error) + Watch(label, field labels.Selector, resourceVersion string) (watch.Interface, error) } // SourceAPI implements a configuration source for services and endpoints that // uses the client watch API to efficiently detect changes. type SourceAPI struct { - client Watcher + servicesWatcher ServicesWatcher + endpointsWatcher EndpointsWatcher + services chan<- ServiceUpdate endpoints chan<- EndpointsUpdate @@ -47,11 +53,12 @@ type SourceAPI struct { } // NewSourceAPI creates a config source that watches for changes to the services and endpoints. -func NewSourceAPI(client Watcher, period time.Duration, services chan<- ServiceUpdate, endpoints chan<- EndpointsUpdate) *SourceAPI { +func NewSourceAPI(servicesWatcher ServicesWatcher, endpointsWatcher EndpointsWatcher, period time.Duration, services chan<- ServiceUpdate, endpoints chan<- EndpointsUpdate) *SourceAPI { config := &SourceAPI{ - client: client, - services: services, - endpoints: endpoints, + servicesWatcher: servicesWatcher, + endpointsWatcher: endpointsWatcher, + services: services, + endpoints: endpoints, waitDuration: period, // prevent hot loops if the server starts to misbehave @@ -72,9 +79,8 @@ func NewSourceAPI(client Watcher, period time.Duration, services chan<- ServiceU // runServices loops forever looking for changes to services. func (s *SourceAPI) runServices(resourceVersion *string) { - ctx := api.NewContext() if len(*resourceVersion) == 0 { - services, err := s.client.ListServices(ctx, labels.Everything()) + services, err := s.servicesWatcher.List(labels.Everything()) if err != nil { glog.Errorf("Unable to load services: %v", err) time.Sleep(wait.Jitter(s.waitDuration, 0.0)) @@ -84,7 +90,7 @@ func (s *SourceAPI) runServices(resourceVersion *string) { s.services <- ServiceUpdate{Op: SET, Services: services.Items} } - watcher, err := s.client.WatchServices(ctx, labels.Everything(), labels.Everything(), *resourceVersion) + watcher, err := s.servicesWatcher.Watch(labels.Everything(), labels.Everything(), *resourceVersion) if err != nil { glog.Errorf("Unable to watch for services changes: %v", err) time.Sleep(wait.Jitter(s.waitDuration, 0.0)) @@ -122,9 +128,8 @@ func handleServicesWatch(resourceVersion *string, ch <-chan watch.Event, updates // runEndpoints loops forever looking for changes to endpoints. func (s *SourceAPI) runEndpoints(resourceVersion *string) { - ctx := api.NewContext() if len(*resourceVersion) == 0 { - endpoints, err := s.client.ListEndpoints(ctx, labels.Everything()) + endpoints, err := s.endpointsWatcher.List(labels.Everything()) if err != nil { glog.Errorf("Unable to load endpoints: %v", err) time.Sleep(wait.Jitter(s.waitDuration, 0.0)) @@ -134,7 +139,7 @@ func (s *SourceAPI) runEndpoints(resourceVersion *string) { s.endpoints <- EndpointsUpdate{Op: SET, Endpoints: endpoints.Items} } - watcher, err := s.client.WatchEndpoints(ctx, labels.Everything(), labels.Everything(), *resourceVersion) + watcher, err := s.endpointsWatcher.Watch(labels.Everything(), labels.Everything(), *resourceVersion) if err != nil { glog.Errorf("Unable to watch for endpoints changes: %v", err) time.Sleep(wait.Jitter(s.waitDuration, 0.0)) diff --git a/pkg/registry/pod/rest.go b/pkg/registry/pod/rest.go index f6715a768b..be6ae5c257 100644 --- a/pkg/registry/pod/rest.go +++ b/pkg/registry/pod/rest.go @@ -273,7 +273,7 @@ func getPodStatus(pod *api.Pod, minions client.MinionInterface) (api.PodStatus, return api.PodWaiting, nil } if minions != nil { - res, err := minions.ListMinions() + res, err := minions.List() if err != nil { glog.Errorf("Error listing minions: %v", err) return "", err diff --git a/pkg/scheduler/predicates.go b/pkg/scheduler/predicates.go index 9f3b15fa60..7c51407e7c 100644 --- a/pkg/scheduler/predicates.go +++ b/pkg/scheduler/predicates.go @@ -48,7 +48,7 @@ type ClientNodeInfo struct { } func (nodes ClientNodeInfo) GetNodeInfo(nodeID string) (*api.Minion, error) { - return nodes.GetMinion(nodeID) + return nodes.Minions().Get(nodeID) } func isVolumeConflict(volume api.Volume, pod *api.Pod) bool { diff --git a/pkg/service/endpoints_controller.go b/pkg/service/endpoints_controller.go index ee908bda78..7c17fa704b 100644 --- a/pkg/service/endpoints_controller.go +++ b/pkg/service/endpoints_controller.go @@ -44,16 +44,14 @@ func NewEndpointController(client *client.Client) *EndpointController { // SyncServiceEndpoints syncs service endpoints. func (e *EndpointController) SyncServiceEndpoints() error { - ctx := api.NewContext() - services, err := e.client.ListServices(ctx, labels.Everything()) + services, err := e.client.Services(api.NamespaceAll).List(labels.Everything()) if err != nil { glog.Errorf("Failed to list services: %v", err) return err } var resultErr error for _, service := range services.Items { - nsCtx := api.WithNamespace(ctx, service.Namespace) - pods, err := e.client.ListPods(nsCtx, labels.Set(service.Selector).AsSelector()) + pods, err := e.client.Pods(service.Namespace).List(labels.Set(service.Selector).AsSelector()) if err != nil { glog.Errorf("Error syncing service: %#v, skipping.", service) resultErr = err @@ -72,7 +70,7 @@ func (e *EndpointController) SyncServiceEndpoints() error { } endpoints = append(endpoints, net.JoinHostPort(pod.CurrentState.PodIP, strconv.Itoa(port))) } - currentEndpoints, err := e.client.GetEndpoints(nsCtx, service.Name) + currentEndpoints, err := e.client.Endpoints(service.Namespace).Get(service.Namespace) if err != nil { if errors.IsNotFound(err) { currentEndpoints = &api.Endpoints{ @@ -91,14 +89,14 @@ func (e *EndpointController) SyncServiceEndpoints() error { if len(currentEndpoints.ResourceVersion) == 0 { // No previous endpoints, create them - _, err = e.client.CreateEndpoints(nsCtx, newEndpoints) + _, err = e.client.Endpoints(service.Namespace).Create(newEndpoints) } else { // Pre-existing if endpointsEqual(currentEndpoints, endpoints) { glog.V(2).Infof("endpoints are equal for %s, skipping update", service.Name) continue } - _, err = e.client.UpdateEndpoints(nsCtx, newEndpoints) + _, err = e.client.Endpoints(service.Namespace).Update(newEndpoints) } if err != nil { glog.Errorf("Error updating endpoints: %#v", err) diff --git a/plugin/cmd/scheduler/scheduler.go b/plugin/cmd/scheduler/scheduler.go index 1fc0927e09..44c364fc4e 100644 --- a/plugin/cmd/scheduler/scheduler.go +++ b/plugin/cmd/scheduler/scheduler.go @@ -56,7 +56,7 @@ func main() { glog.Fatalf("Invalid API configuration: %v", err) } - record.StartRecording(kubeClient, "scheduler") + record.StartRecording(kubeClient.Events(), "scheduler") go http.ListenAndServe(net.JoinHostPort(address.String(), strconv.Itoa(*port)), nil)