github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-30 06:54:01 +00:00
Code Issues Projects Releases Wiki Activity

Move Resource functionality to its own package

Browse Source 
Create a unified Builder object for working with files, selectors,
types, and items that makes it easier to get multi-object functionality.

Supports all of the behaviors previously in resource.go, but with
additional flexibility to allow multi-type retrieval and access, directories,
URLs, nested objects, and lists.
This commit is contained in:
Clayton Coleman 2014-12-27 16:48:27 -05:00
parent 68298f08a4
commit d75a3d5021
10 changed files with 2487 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
pkg/kubectl/cmd/helpers.go
View File

@ -105,6 +105,7 @@ func FirstNonEmptyString(args ...string) string {
}
// Return a list of file names of a certain type within a given directory.
// TODO: replace with resource.Builder
func GetFilesFromDir(directory string, fileType string) []string {
files := []string{}
@ -121,6 +122,7 @@ func GetFilesFromDir(directory string, fileType string) []string {
// ReadConfigData reads the bytes from the specified filesytem or network
// location or from stdin if location == "-".
// TODO: replace with resource.Builder
func ReadConfigData(location string) ([]byte, error) {
if len(location) == 0 {
return nil, fmt.Errorf("location given but empty")
@ -144,6 +146,7 @@ func ReadConfigData(location string) ([]byte, error) {
return ReadConfigDataFromLocation(location)
}
// TODO: replace with resource.Builder
func ReadConfigDataFromLocation(location string) ([]byte, error) {
// we look for http:// or https:// to determine if valid URL, otherwise do normal file IO
if strings.Index(location, "http://") == 0 || strings.Index(location, "https://") == 0 {

563
pkg/kubectl/resource/builder.go Normal file
View File

@ -0,0 +1,563 @@
/*
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 resource
import (
"fmt"
"io"
"net/url"
"os"
"reflect"
"strings"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/errors"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
)
// Builder provides convenience functions for taking arguments and parameters
// from the command line and converting them to a list of resources to iterate
// over using the Visitor interface.
type Builder struct {
mapper *Mapper
errs []error
paths []Visitor
stream bool
dir bool
selector labels.Selector
resources []string
namespace string
name string
defaultNamespace bool
requireNamespace bool
flatten bool
latest bool
singleResourceType bool
continueOnError bool
}
// NewBuilder creates a builder that operates on generic objects.
func NewBuilder(mapper meta.RESTMapper, typer runtime.ObjectTyper, clientMapper ClientMapper) *Builder {
return &Builder{
mapper: &Mapper{typer, mapper, clientMapper},
}
}
// Filename is parameters passed via a filename argument which may be URLs, the "-" argument indicating
// STDIN, or paths to files or directories. If ContinueOnError() is set prior to this method being called,
// objects on the path that are unrecognized will be ignored (but logged at V(2)).
func (b *Builder) FilenameParam(paths ...string) *Builder {
for _, s := range paths {
switch {
case s == "-":
b.Stdin()
case strings.Index(s, "http://") == 0 || strings.Index(s, "https://") == 0:
url, err := url.Parse(s)
if err != nil {
b.errs = append(b.errs, fmt.Errorf("the URL passed to filename %q is not valid: %v", s, err))
continue
}
b.URL(url)
default:
b.Path(s)
}
}
return b
}
// URL accepts a number of URLs directly.
func (b *Builder) URL(urls ...*url.URL) *Builder {
for _, u := range urls {
b.paths = append(b.paths, &URLVisitor{
Mapper: b.mapper,
URL: u,
})
}
return b
}
// Stdin will read objects from the standard input. If ContinueOnError() is set
// prior to this method being called, objects in the stream that are unrecognized
// will be ignored (but logged at V(2)).
func (b *Builder) Stdin() *Builder {
return b.Stream(os.Stdin, "STDIN")
}
// Stream will read objects from the provided reader, and if an error occurs will
// include the name string in the error message. If ContinueOnError() is set
// prior to this method being called, objects in the stream that are unrecognized
// will be ignored (but logged at V(2)).
func (b *Builder) Stream(r io.Reader, name string) *Builder {
b.stream = true
b.paths = append(b.paths, NewStreamVisitor(r, b.mapper, name, b.continueOnError))
return b
}
// Path is a set of filesystem paths that may be files containing one or more
// resources. If ContinueOnError() is set prior to this method being called,
// objects on the path that are unrecognized will be ignored (but logged at V(2)).
func (b *Builder) Path(paths ...string) *Builder {
for _, p := range paths {
i, err := os.Stat(p)
if os.IsNotExist(err) {
b.errs = append(b.errs, fmt.Errorf("the path %q does not exist", p))
continue
}
if err != nil {
b.errs = append(b.errs, fmt.Errorf("the path %q cannot be accessed: %v", p, err))
continue
}
var visitor Visitor
if i.IsDir() {
b.dir = true
visitor = &DirectoryVisitor{
Mapper: b.mapper,
Path: p,
Extensions: []string{".json", ".yaml"},
Recursive: false,
IgnoreErrors: b.continueOnError,
}
} else {
visitor = &PathVisitor{
Mapper: b.mapper,
Path: p,
IgnoreErrors: b.continueOnError,
}
}
b.paths = append(b.paths, visitor)
}
return b
}
// ResourceTypes is a list of types of resources to operate on, when listing objects on
// the server or retrieving objects that match a selector.
func (b *Builder) ResourceTypes(types ...string) *Builder {
b.resources = append(b.resources, types...)
return b
}
// SelectorParam defines a selector that should be applied to the object types to load.
// This will not affect files loaded from disk or URL. If the parameter is empty it is
// a no-op - to select all resources invoke `b.Selector(labels.Everything)`.
func (b *Builder) SelectorParam(s string) *Builder {
selector, err := labels.ParseSelector(s)
if err != nil {
b.errs = append(b.errs, fmt.Errorf("the provided selector %q is not valid: %v", s, err))
}
if selector.Empty() {
return b
}
return b.Selector(selector)
}
// Selector accepts a selector directly, and if non nil will trigger a list action.
func (b *Builder) Selector(selector labels.Selector) *Builder {
b.selector = selector
return b
}
// The namespace that these resources should be assumed to under - used by DefaultNamespace()
// and RequireNamespace()
func (b *Builder) NamespaceParam(namespace string) *Builder {
b.namespace = namespace
return b
}
// DefaultNamespace instructs the builder to set the namespace value for any object found
// to NamespaceParam() if empty.
func (b *Builder) DefaultNamespace() *Builder {
b.defaultNamespace = true
return b
}
// RequireNamespace instructs the builder to set the namespace value for any object found
// to NamespaceParam() if empty, and if the value on the resource does not match
// NamespaceParam() an error will be returned.
func (b *Builder) RequireNamespace() *Builder {
b.requireNamespace = true
return b
}
// ResourceTypeOrNameArgs indicates that the builder should accept one or two arguments
// of the form `(<type1>[,<type2>,...]|<type> <name>)`. When one argument is received, the types
// provided will be retrieved from the server (and be comma delimited). When two arguments are
// received, they must be a single type and name. If more than two arguments are provided an
// error is set.
func (b *Builder) ResourceTypeOrNameArgs(args ...string) *Builder {
switch len(args) {
case 2:
b.name = args[1]
b.ResourceTypes(SplitResourceArgument(args[0])...)
case 1:
b.ResourceTypes(SplitResourceArgument(args[0])...)
if b.selector == nil {
b.selector = labels.Everything()
}
case 0:
default:
b.errs = append(b.errs, fmt.Errorf("when passing arguments, must be resource or resource and name"))
}
return b
}
// ResourceTypeAndNameArgs expects two arguments, a resource type, and a resource name. The resource
// matching that type and and name will be retrieved from the server.
func (b *Builder) ResourceTypeAndNameArgs(args ...string) *Builder {
switch len(args) {
case 2:
b.name = args[1]
b.ResourceTypes(SplitResourceArgument(args[0])...)
case 0:
default:
b.errs = append(b.errs, fmt.Errorf("when passing arguments, must be resource and name"))
}
return b
}
// Flatten will convert any objects with a field named "Items" that is an array of runtime.Object
// compatible types into individual entries and give them their own items. The original object
// is not passed to any visitors.
func (b *Builder) Flatten() *Builder {
b.flatten = true
return b
}
// Latest will fetch the latest copy of any objects loaded from URLs or files from the server.
func (b *Builder) Latest() *Builder {
b.latest = true
return b
}
// ContinueOnError will attempt to load and visit as many objects as possible, even if some visits
// return errors or some objects cannot be loaded. The default behavior is to terminate after
// the first error is returned from a VisitorFunc.
func (b *Builder) ContinueOnError() *Builder {
b.continueOnError = true
return b
}
func (b *Builder) SingleResourceType() *Builder {
b.singleResourceType = true
return b
}
func (b *Builder) resourceMappings() ([]*meta.RESTMapping, error) {
if len(b.resources) > 1 && b.singleResourceType {
return nil, fmt.Errorf("you may only specify a single resource type")
}
mappings := []*meta.RESTMapping{}
for _, r := range b.resources {
version, kind, err := b.mapper.VersionAndKindForResource(r)
if err != nil {
return nil, err
}
mapping, err := b.mapper.RESTMapping(kind, version)
if err != nil {
return nil, err
}
mappings = append(mappings, mapping)
}
return mappings, nil
}
func (b *Builder) visitorResult() *Result {
if len(b.errs) > 0 {
return &Result{err: errors.NewAggregate(b.errs)}
}
// visit selectors
if b.selector != nil {
if len(b.name) != 0 {
return &Result{err: fmt.Errorf("name cannot be provided when a selector is specified")}
}
if len(b.resources) == 0 {
return &Result{err: fmt.Errorf("at least one resource must be specified to use a selector")}
}
// empty selector has different error message for paths being provided
if len(b.paths) != 0 {
if b.selector.Empty() {
return &Result{err: fmt.Errorf("when paths, URLs, or stdin is provided as input, you may not specify a resource by arguments as well")}
} else {
return &Result{err: fmt.Errorf("a selector may not be specified when path, URL, or stdin is provided as input")}
}
}
mappings, err := b.resourceMappings()
if err != nil {
return &Result{err: err}
}
visitors := []Visitor{}
for _, mapping := range mappings {
client, err := b.mapper.ClientForMapping(mapping)
if err != nil {
return &Result{err: err}
}
visitors = append(visitors, NewSelector(client, mapping, b.namespace, b.selector))
}
if b.continueOnError {
return &Result{visitor: EagerVisitorList(visitors), sources: visitors}
}
return &Result{visitor: VisitorList(visitors), sources: visitors}
}
// visit single item specified by name
if len(b.name) != 0 {
if len(b.paths) != 0 {
return &Result{singular: true, err: fmt.Errorf("when paths, URLs, or stdin is provided as input, you may not specify a resource by arguments as well")}
}
if len(b.resources) == 0 {
return &Result{singular: true, err: fmt.Errorf("you must provide a resource and a resource name together")}
}
if len(b.resources) > 1 {
return &Result{singular: true, err: fmt.Errorf("you must specify only one resource")}
}
if len(b.namespace) == 0 {
return &Result{singular: true, err: fmt.Errorf("namespace may not be empty when retrieving a resource by name")}
}
mappings, err := b.resourceMappings()
if err != nil {
return &Result{singular: true, err: err}
}
client, err := b.mapper.ClientForMapping(mappings[0])
if err != nil {
return &Result{singular: true, err: err}
}
info := NewInfo(client, mappings[0], b.namespace, b.name)
if err := info.Get(); err != nil {
return &Result{singular: true, err: err}
}
return &Result{singular: true, visitor: info, sources: []Visitor{info}}
}
// visit items specified by paths
if len(b.paths) != 0 {
singular := !b.dir && !b.stream && len(b.paths) == 1
if len(b.resources) != 0 {
return &Result{singular: singular, err: fmt.Errorf("when paths, URLs, or stdin is provided as input, you may not specify resource arguments as well")}
}
var visitors Visitor
if b.continueOnError {
visitors = EagerVisitorList(b.paths)
} else {
visitors = VisitorList(b.paths)
}
// only items from disk can be refetched
if b.latest {
// must flatten lists prior to fetching
if b.flatten {
visitors = NewFlattenListVisitor(visitors, b.mapper)
}
visitors = NewDecoratedVisitor(visitors, RetrieveLatest)
}
return &Result{singular: singular, visitor: visitors, sources: b.paths}
}
return &Result{err: fmt.Errorf("you must provide one or more resources by argument or filename")}
}
// Do returns a Result object with a Visitor for the resources identified by the Builder.
// The visitor will respect the error behavior specified by ContinueOnError. Note that stream
// inputs are consumed by the first execution - use Infos() or Object() on the Result to capture a list
// for further iteration.
func (b *Builder) Do() *Result {
r := b.visitorResult()
if r.err != nil {
return r
}
if b.flatten {
r.visitor = NewFlattenListVisitor(r.visitor, b.mapper)
}
helpers := []VisitorFunc{}
if b.defaultNamespace {
helpers = append(helpers, SetNamespace(b.namespace))
}
if b.requireNamespace {
helpers = append(helpers, RequireNamespace(b.namespace))
}
r.visitor = NewDecoratedVisitor(r.visitor, helpers...)
return r
}
// Result contains helper methods for dealing with the outcome of a Builder.
type Result struct {
err error
visitor Visitor
sources []Visitor
singular bool
// populated by a call to Infos
info []*Info
}
// Err returns one or more errors (via a util.ErrorList) that occurred prior
// to visiting the elements in the visitor. To see all errors including those
// that occur during visitation, invoke Infos().
func (r *Result) Err() error {
return r.err
}
// Visit implements the Visitor interface on the items described in the Builder.
// Note that some visitor sources are not traversable more than once, or may
// return different results. If you wish to operate on the same set of resources
// multiple times, use the Infos() method.
func (r *Result) Visit(fn VisitorFunc) error {
if r.err != nil {
return r.err
}
return r.visitor.Visit(fn)
}
// IntoSingular sets the provided boolean pointer to true if the Builder input
// reflected a single item, or multiple.
func (r *Result) IntoSingular(b *bool) *Result {
*b = r.singular
return r
}
// Infos returns an array of all of the resource infos retrieved via traversal.
// Will attempt to traverse the entire set of visitors only once, and will return
// a cached list on subsequent calls.
func (r *Result) Infos() ([]*Info, error) {
if r.err != nil {
return nil, r.err
}
if r.info != nil {
return r.info, nil
}
infos := []*Info{}
err := r.visitor.Visit(func(info *Info) error {
infos = append(infos, info)
return nil
})
r.info, r.err = infos, err
return infos, err
}
// Object returns a single object representing the output of a single visit to all
// found resources. If the Builder was a singular context (expected to return a
// single resource by user input) and only a single resource was found, the resource
// will be returned as is. Otherwise, the returned resources will be part of an
// api.List. The ResourceVersion of the api.List will be set only if it is identical
// across all infos returned.
func (r *Result) Object() (runtime.Object, error) {
infos, err := r.Infos()
if err != nil {
return nil, err
}
versions := util.StringSet{}
objects := []runtime.Object{}
for _, info := range infos {
if info.Object != nil {
objects = append(objects, info.Object)
versions.Insert(info.ResourceVersion)
}
}
if len(objects) == 1 {
if r.singular {
return objects[0], nil
}
// if the item is a list already, don't create another list
if _, err := runtime.GetItemsPtr(objects[0]); err == nil {
return objects[0], nil
}
}
version := ""
if len(versions) == 1 {
version = versions.List()[0]
}
return &api.List{
ListMeta: api.ListMeta{
ResourceVersion: version,
},
Items: objects,
}, err
}
// ResourceMapping returns a single meta.RESTMapping representing the
// resources located by the builder, or an error if more than one
// mapping was found.
func (r *Result) ResourceMapping() (*meta.RESTMapping, error) {
if r.err != nil {
return nil, r.err
}
mappings := map[string]*meta.RESTMapping{}
for i := range r.sources {
m, ok := r.sources[i].(ResourceMapping)
if !ok {
return nil, fmt.Errorf("a resource mapping could not be loaded from %v", reflect.TypeOf(r.sources[i]))
}
mapping := m.ResourceMapping()
mappings[mapping.Resource] = mapping
}
if len(mappings) != 1 {
return nil, fmt.Errorf("expected only a single resource type")
}
for _, mapping := range mappings {
return mapping, nil
}
return nil, nil
}
// Watch retrieves changes that occur on the server to the specified resource.
// It currently supports watching a single source - if the resource source
// (selectors or pure types) can be watched, they will be, otherwise the list
// will be visited (equivalent to the Infos() call) and if there is a single
// resource present, it will be watched, otherwise an error will be returned.
func (r *Result) Watch(resourceVersion string) (watch.Interface, error) {
if r.err != nil {
return nil, r.err
}
if len(r.sources) != 1 {
return nil, fmt.Errorf("you may only watch a single resource or type of resource at a time")
}
w, ok := r.sources[0].(Watchable)
if !ok {
info, err := r.Infos()
if err != nil {
return nil, err
}
if len(info) != 1 {
return nil, fmt.Errorf("watch is only supported on a single resource - %d resources were found", len(info))
}
return info[0].Watch(resourceVersion)
}
return w.Watch(resourceVersion)
}
func SplitResourceArgument(arg string) []string {
set := util.NewStringSet()
set.Insert(strings.Split(arg, ",")...)
return set.List()
}

619
pkg/kubectl/resource/builder_test.go Normal file
View File

@ -0,0 +1,619 @@
/*
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 resource
import (
"bytes"
"io"
"io/ioutil"
"net/http"
"net/http/httptest"
"reflect"
"testing"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/latest"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/errors"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
watchjson "github.com/GoogleCloudPlatform/kubernetes/pkg/watch/json"
)
func stringBody(body string) io.ReadCloser {
return ioutil.NopCloser(bytes.NewReader([]byte(body)))
}
func watchBody(events ...watch.Event) string {
buf := &bytes.Buffer{}
enc := watchjson.NewEncoder(buf, latest.Codec)
for _, e := range events {
enc.Encode(&e)
}
return buf.String()
}
func fakeClient() ClientMapper {
return ClientMapperFunc(func(*meta.RESTMapping) (RESTClient, error) {
return &client.FakeRESTClient{}, nil
})
}
func fakeClientWith(t *testing.T, data map[string]string) ClientMapper {
return ClientMapperFunc(func(*meta.RESTMapping) (RESTClient, error) {
return &client.FakeRESTClient{
Codec: latest.Codec,
Client: client.HTTPClientFunc(func(req *http.Request) (*http.Response, error) {
p := req.URL.Path
q := req.URL.RawQuery
if len(q) != 0 {
p = p + "?" + q
}
body, ok := data[p]
if !ok {
t.Fatalf("unexpected request: %s (%s)\n%#v", p, req.URL, req)
}
return &http.Response{
StatusCode: http.StatusOK,
Body: stringBody(body),
}, nil
}),
}, nil
})
}
func testData() (*api.PodList, *api.ServiceList) {
pods := &api.PodList{
ListMeta: api.ListMeta{
ResourceVersion: "15",
},
Items: []api.Pod{
{
ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: "test", ResourceVersion: "10"},
},
{
ObjectMeta: api.ObjectMeta{Name: "bar", Namespace: "test", ResourceVersion: "11"},
},
},
}
svc := &api.ServiceList{
ListMeta: api.ListMeta{
ResourceVersion: "16",
},
Items: []api.Service{
{
ObjectMeta: api.ObjectMeta{Name: "baz", Namespace: "test", ResourceVersion: "12"},
},
},
}
return pods, svc
}
func streamTestData() (io.Reader, *api.PodList, *api.ServiceList) {
pods, svc := testData()
r, w := io.Pipe()
go func() {
defer w.Close()
w.Write([]byte(runtime.EncodeOrDie(latest.Codec, pods)))
w.Write([]byte(runtime.EncodeOrDie(latest.Codec, svc)))
}()
return r, pods, svc
}
type testVisitor struct {
InjectErr error
Infos []*Info
}
func (v *testVisitor) Handle(info *Info) error {
v.Infos = append(v.Infos, info)
return v.InjectErr
}
func (v *testVisitor) Objects() []runtime.Object {
objects := []runtime.Object{}
for i := range v.Infos {
objects = append(objects, v.Infos[i].Object)
}
return objects
}
func TestPathBuilder(t *testing.T) {
b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClient()).
FilenameParam("../../../examples/guestbook/redis-master.json")
test := &testVisitor{}
singular := false
err := b.Do().IntoSingular(&singular).Visit(test.Handle)
if err != nil || !singular || len(test.Infos) != 1 {
t.Fatalf("unexpected response: %v %f %#v", err, singular, test.Infos)
}
info := test.Infos[0]
if info.Name != "redis-master" || info.Namespace != "" || info.Object == nil {
t.Errorf("unexpected info: %#v", info)
}
}
func TestPathBuilderWithMultiple(t *testing.T) {
b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClient()).
FilenameParam("../../../examples/guestbook/redis-master.json").
FilenameParam("../../../examples/guestbook/redis-master.json").
NamespaceParam("test").DefaultNamespace()
test := &testVisitor{}
singular := false
err := b.Do().IntoSingular(&singular).Visit(test.Handle)
if err != nil || singular || len(test.Infos) != 2 {
t.Fatalf("unexpected response: %v %f %#v", err, singular, test.Infos)
}
info := test.Infos[1]
if info.Name != "redis-master" || info.Namespace != "test" || info.Object == nil {
t.Errorf("unexpected info: %#v", info)
}
}
func TestDirectoryBuilder(t *testing.T) {
b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClient()).
FilenameParam("../../../examples/guestbook").
NamespaceParam("test").DefaultNamespace()
test := &testVisitor{}
singular := false
err := b.Do().IntoSingular(&singular).Visit(test.Handle)
if err != nil || singular || len(test.Infos) < 4 {
t.Fatalf("unexpected response: %v %f %#v", err, singular, test.Infos)
}
found := false
for _, info := range test.Infos {
if info.Name == "redis-master" && info.Namespace == "test" && info.Object != nil {
found = true
}
}
if !found {
t.Errorf("unexpected responses: %#v", test.Infos)
}
}
func TestURLBuilder(t *testing.T) {
s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
w.WriteHeader(http.StatusOK)
w.Write([]byte(runtime.EncodeOrDie(latest.Codec, &api.Pod{ObjectMeta: api.ObjectMeta{Namespace: "foo", Name: "test"}})))
}))
defer s.Close()
b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClient()).
FilenameParam(s.URL).
NamespaceParam("test")
test := &testVisitor{}
singular := false
err := b.Do().IntoSingular(&singular).Visit(test.Handle)
if err != nil || !singular || len(test.Infos) != 1 {
t.Fatalf("unexpected response: %v %f %#v", err, singular, test.Infos)
}
info := test.Infos[0]
if info.Name != "test" || info.Namespace != "foo" || info.Object == nil {
t.Errorf("unexpected info: %#v", info)
}
}
func TestURLBuilderRequireNamespace(t *testing.T) {
s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
w.WriteHeader(http.StatusOK)
w.Write([]byte(runtime.EncodeOrDie(latest.Codec, &api.Pod{ObjectMeta: api.ObjectMeta{Namespace: "foo", Name: "test"}})))
}))
defer s.Close()
b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClient()).
FilenameParam(s.URL).
NamespaceParam("test").RequireNamespace()
test := &testVisitor{}
singular := false
err := b.Do().IntoSingular(&singular).Visit(test.Handle)
if err == nil || !singular || len(test.Infos) != 0 {
t.Fatalf("unexpected response: %v %f %#v", err, singular, test.Infos)
}
}
func TestResourceByName(t *testing.T) {
pods, _ := testData()
b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClientWith(t, map[string]string{
"/ns/test/pods/foo": runtime.EncodeOrDie(latest.Codec, &pods.Items[0]),
})).
NamespaceParam("test")
test := &testVisitor{}
singular := false
if b.Do().Err() == nil {
t.Errorf("unexpected non-error")
}
b.ResourceTypeOrNameArgs("pods", "foo")
err := b.Do().IntoSingular(&singular).Visit(test.Handle)
if err != nil || !singular || len(test.Infos) != 1 {
t.Fatalf("unexpected response: %v %f %#v", err, singular, test.Infos)
}
if !reflect.DeepEqual(&pods.Items[0], test.Objects()[0]) {
t.Errorf("unexpected object: %#v", test.Objects())
}
mapping, err := b.Do().ResourceMapping()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if mapping.Resource != "pods" {
t.Errorf("unexpected resource mapping: %#v", mapping)
}
}
func TestResourceByNameAndEmptySelector(t *testing.T) {
pods, _ := testData()
b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClientWith(t, map[string]string{
"/ns/test/pods/foo": runtime.EncodeOrDie(latest.Codec, &pods.Items[0]),
})).
NamespaceParam("test").
SelectorParam("").
ResourceTypeOrNameArgs("pods", "foo")
singular := false
infos, err := b.Do().IntoSingular(&singular).Infos()
if err != nil || !singular || len(infos) != 1 {
t.Fatalf("unexpected response: %v %f %#v", err, singular, infos)
}
if !reflect.DeepEqual(&pods.Items[0], infos[0].Object) {
t.Errorf("unexpected object: %#v", infos[0])
}
mapping, err := b.Do().ResourceMapping()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if mapping.Resource != "pods" {
t.Errorf("unexpected resource mapping: %#v", mapping)
}
}
func TestSelector(t *testing.T) {
pods, svc := testData()
b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClientWith(t, map[string]string{
"/ns/test/pods?labels=a%3Db": runtime.EncodeOrDie(latest.Codec, pods),
"/ns/test/services?labels=a%3Db": runtime.EncodeOrDie(latest.Codec, svc),
})).
SelectorParam("a=b").
NamespaceParam("test").
Flatten()
test := &testVisitor{}
singular := false
if b.Do().Err() == nil {
t.Errorf("unexpected non-error")
}
b.ResourceTypeOrNameArgs("pods,service")
err := b.Do().IntoSingular(&singular).Visit(test.Handle)
if err != nil || singular || len(test.Infos) != 3 {
t.Fatalf("unexpected response: %v %f %#v", err, singular, test.Infos)
}
if !reflect.DeepEqual([]runtime.Object{&pods.Items[0], &pods.Items[1], &svc.Items[0]}, test.Objects()) {
t.Errorf("unexpected visited objects: %#v", test.Objects())
}
if _, err := b.Do().ResourceMapping(); err == nil {
t.Errorf("unexpected non-error")
}
}
func TestSelectorRequiresKnownTypes(t *testing.T) {
b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClient()).
SelectorParam("a=b").
NamespaceParam("test").
ResourceTypes("unknown")
if b.Do().Err() == nil {
t.Errorf("unexpected non-error")
}
}
func TestSingleResourceType(t *testing.T) {
b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClient()).
SelectorParam("a=b").
SingleResourceType().
ResourceTypeOrNameArgs("pods,services")
if b.Do().Err() == nil {
t.Errorf("unexpected non-error")
}
}
func TestStream(t *testing.T) {
r, pods, rc := streamTestData()
b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClient()).
NamespaceParam("test").Stream(r, "STDIN").Flatten()
test := &testVisitor{}
singular := false
err := b.Do().IntoSingular(&singular).Visit(test.Handle)
if err != nil || singular || len(test.Infos) != 3 {
t.Fatalf("unexpected response: %v %f %#v", err, singular, test.Infos)
}
if !reflect.DeepEqual([]runtime.Object{&pods.Items[0], &pods.Items[1], &rc.Items[0]}, test.Objects()) {
t.Errorf("unexpected visited objects: %#v", test.Objects())
}
}
func TestMultipleObject(t *testing.T) {
r, pods, svc := streamTestData()
obj, err := NewBuilder(latest.RESTMapper, api.Scheme, fakeClient()).
NamespaceParam("test").Stream(r, "STDIN").Flatten().
Do().Object()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
expected := &api.List{
Items: []runtime.Object{
&pods.Items[0],
&pods.Items[1],
&svc.Items[0],
},
}
if !reflect.DeepEqual(expected, obj) {
t.Errorf("unexpected visited objects: %#v", obj)
}
}
func TestSingularObject(t *testing.T) {
obj, err := NewBuilder(latest.RESTMapper, api.Scheme, fakeClient()).
NamespaceParam("test").DefaultNamespace().
FilenameParam("../../../examples/guestbook/redis-master.json").
Flatten().
Do().Object()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
pod, ok := obj.(*api.Pod)
if !ok {
t.Fatalf("unexpected object: %#v", obj)
}
if pod.Name != "redis-master" || pod.Namespace != "test" {
t.Errorf("unexpected pod: %#v", pod)
}
}
func TestListObject(t *testing.T) {
pods, _ := testData()
b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClientWith(t, map[string]string{
"/ns/test/pods?labels=a%3Db": runtime.EncodeOrDie(latest.Codec, pods),
})).
SelectorParam("a=b").
NamespaceParam("test").
ResourceTypeOrNameArgs("pods").
Flatten()
obj, err := b.Do().Object()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
list, ok := obj.(*api.List)
if !ok {
t.Fatalf("unexpected object: %#v", obj)
}
if list.ResourceVersion != pods.ResourceVersion || len(list.Items) != 2 {
t.Errorf("unexpected list: %#v", list)
}
mapping, err := b.Do().ResourceMapping()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if mapping.Resource != "pods" {
t.Errorf("unexpected resource mapping: %#v", mapping)
}
}
func TestListObjectWithDifferentVersions(t *testing.T) {
pods, svc := testData()
obj, err := NewBuilder(latest.RESTMapper, api.Scheme, fakeClientWith(t, map[string]string{
"/ns/test/pods?labels=a%3Db": runtime.EncodeOrDie(latest.Codec, pods),
"/ns/test/services?labels=a%3Db": runtime.EncodeOrDie(latest.Codec, svc),
})).
SelectorParam("a=b").
NamespaceParam("test").
ResourceTypeOrNameArgs("pods,services").
Flatten().
Do().Object()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
list, ok := obj.(*api.List)
if !ok {
t.Fatalf("unexpected object: %#v", obj)
}
// resource version differs between type lists, so it's not possible to get a single version.
if list.ResourceVersion != "" || len(list.Items) != 3 {
t.Errorf("unexpected list: %#v", list)
}
}
func TestWatch(t *testing.T) {
pods, _ := testData()
w, err := NewBuilder(latest.RESTMapper, api.Scheme, fakeClientWith(t, map[string]string{
"/watch/ns/test/pods/redis-master?resourceVersion=10": watchBody(watch.Event{
Type: watch.Added,
Object: &pods.Items[0],
}),
})).
NamespaceParam("test").DefaultNamespace().
FilenameParam("../../../examples/guestbook/redis-master.json").Flatten().
Do().Watch("10")
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
defer w.Stop()
ch := w.ResultChan()
select {
case obj := <-ch:
if obj.Type != watch.Added {
t.Fatalf("unexpected watch event", obj)
}
pod, ok := obj.Object.(*api.Pod)
if !ok {
t.Fatalf("unexpected object: %#v", obj)
}
if pod.Name != "foo" || pod.ResourceVersion != "10" {
t.Errorf("unexpected pod: %#v", pod)
}
}
}
func TestWatchMultipleError(t *testing.T) {
_, err := NewBuilder(latest.RESTMapper, api.Scheme, fakeClient()).
NamespaceParam("test").DefaultNamespace().
FilenameParam("../../../examples/guestbook/redis-master.json").Flatten().
FilenameParam("../../../examples/guestbook/redis-master.json").Flatten().
Do().Watch("")
if err == nil {
t.Fatalf("unexpected non-error")
}
}
func TestLatest(t *testing.T) {
r, _, _ := streamTestData()
newPod := &api.Pod{
ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: "test", ResourceVersion: "13"},
}
newPod2 := &api.Pod{
ObjectMeta: api.ObjectMeta{Name: "bar", Namespace: "test", ResourceVersion: "14"},
}
newSvc := &api.Service{
ObjectMeta: api.ObjectMeta{Name: "baz", Namespace: "test", ResourceVersion: "15"},
}
b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClientWith(t, map[string]string{
"/ns/test/pods/foo": runtime.EncodeOrDie(latest.Codec, newPod),
"/ns/test/pods/bar": runtime.EncodeOrDie(latest.Codec, newPod2),
"/ns/test/services/baz": runtime.EncodeOrDie(latest.Codec, newSvc),
})).
NamespaceParam("other").Stream(r, "STDIN").Flatten().Latest()
test := &testVisitor{}
singular := false
err := b.Do().IntoSingular(&singular).Visit(test.Handle)
if err != nil || singular || len(test.Infos) != 3 {
t.Fatalf("unexpected response: %v %f %#v", err, singular, test.Infos)
}
if !reflect.DeepEqual([]runtime.Object{newPod, newPod2, newSvc}, test.Objects()) {
t.Errorf("unexpected visited objects: %#v", test.Objects())
}
}
func TestIgnoreStreamErrors(t *testing.T) {
pods, svc := testData()
r, w := io.Pipe()
go func() {
defer w.Close()
w.Write([]byte(`{}`))
w.Write([]byte(runtime.EncodeOrDie(latest.Codec, &pods.Items[0])))
}()
r2, w2 := io.Pipe()
go func() {
defer w2.Close()
w2.Write([]byte(`{}`))
w2.Write([]byte(runtime.EncodeOrDie(latest.Codec, &svc.Items[0])))
}()
b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClient()).
ContinueOnError(). // TODO: order seems bad, but allows clients to determine what they want...
Stream(r, "1").Stream(r2, "2")
test := &testVisitor{}
singular := false
err := b.Do().IntoSingular(&singular).Visit(test.Handle)
if err != nil || singular || len(test.Infos) != 2 {
t.Fatalf("unexpected response: %v %f %#v", err, singular, test.Infos)
}
if !reflect.DeepEqual([]runtime.Object{&pods.Items[0], &svc.Items[0]}, test.Objects()) {
t.Errorf("unexpected visited objects: %#v", test.Objects())
}
}
func TestReceiveMultipleErrors(t *testing.T) {
pods, svc := testData()
r, w := io.Pipe()
go func() {
defer w.Close()
w.Write([]byte(`{}`))
w.Write([]byte(runtime.EncodeOrDie(latest.Codec, &pods.Items[0])))
}()
r2, w2 := io.Pipe()
go func() {
defer w2.Close()
w2.Write([]byte(`{}`))
w2.Write([]byte(runtime.EncodeOrDie(latest.Codec, &svc.Items[0])))
}()
b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClient()).
Stream(r, "1").Stream(r2, "2").
ContinueOnError()
test := &testVisitor{}
singular := false
err := b.Do().IntoSingular(&singular).Visit(test.Handle)
if err == nil || singular || len(test.Infos) != 0 {
t.Fatalf("unexpected response: %v %f %#v", err, singular, test.Infos)
}
errs, ok := err.(errors.Aggregate)
if !ok {
t.Fatalf("unexpected error: %v", reflect.TypeOf(err))
}
if len(errs.Errors()) != 2 {
t.Errorf("unexpected errors", errs)
}
}

24
pkg/kubectl/resource/doc.go Normal file
View File

@ -0,0 +1,24 @@
/*
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 resource assists clients in dealing with RESTful objects that match the
// Kubernetes API conventions. The Helper object provides simple CRUD operations
// on resources. The Visitor interface makes it easy to deal with multiple resources
// in bulk for retrieval and operation. The Builder object simplifies converting
// standard command line arguments and parameters into a Visitor that can iterate
// over all of the identified resources, whether on the server or on the local
// filesystem.
package resource

172
pkg/kubectl/resource/helper.go Normal file
View File

@ -0,0 +1,172 @@
/*
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 resource
import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
)
// Helper provides methods for retrieving or mutating a RESTful
// resource.
type Helper struct {
// The name of this resource as the server would recognize it
Resource string
// A RESTClient capable of mutating this resource
RESTClient RESTClient
// A codec for decoding and encoding objects of this resource type.
Codec runtime.Codec
// An interface for reading or writing the resource version of this
// type.
Versioner runtime.ResourceVersioner
}
// NewHelper creates a Helper from a ResourceMapping
func NewHelper(client RESTClient, mapping *meta.RESTMapping) *Helper {
return &Helper{
RESTClient: client,
Resource: mapping.Resource,
Codec: mapping.Codec,
Versioner: mapping.MetadataAccessor,
}
}
func (m *Helper) Get(namespace, name string) (runtime.Object, error) {
return m.RESTClient.Get().
Namespace(namespace).
Resource(m.Resource).
Name(name).
Do().
Get()
}
func (m *Helper) List(namespace string, selector labels.Selector) (runtime.Object, error) {
return m.RESTClient.Get().
Namespace(namespace).
Resource(m.Resource).
SelectorParam("labels", selector).
Do().
Get()
}
func (m *Helper) Watch(namespace, resourceVersion string, labelSelector, fieldSelector labels.Selector) (watch.Interface, error) {
return m.RESTClient.Get().
Prefix("watch").
Namespace(namespace).
Resource(m.Resource).
Param("resourceVersion", resourceVersion).
SelectorParam("labels", labelSelector).
SelectorParam("fields", fieldSelector).
Watch()
}
func (m *Helper) WatchSingle(namespace, name, resourceVersion string) (watch.Interface, error) {
return m.RESTClient.Get().
Prefix("watch").
Namespace(namespace).
Resource(m.Resource).
Name(name).
Param("resourceVersion", resourceVersion).
Watch()
}
func (m *Helper) Delete(namespace, name string) error {
return m.RESTClient.Delete().
Namespace(namespace).
Resource(m.Resource).
Name(name).
Do().
Error()
}
func (m *Helper) Create(namespace string, modify bool, data []byte) error {
if modify {
obj, err := m.Codec.Decode(data)
if err != nil {
// We don't know how to check a version on this object, but create it anyway
return createResource(m.RESTClient, m.Resource, namespace, data)
}
// Attempt to version the object based on client logic.
version, err := m.Versioner.ResourceVersion(obj)
if err != nil {
// We don't know how to clear the version on this object, so send it to the server as is
return createResource(m.RESTClient, m.Resource, namespace, data)
}
if version != "" {
if err := m.Versioner.SetResourceVersion(obj, ""); err != nil {
return err
}
newData, err := m.Codec.Encode(obj)
if err != nil {
return err
}
data = newData
}
}
return createResource(m.RESTClient, m.Resource, namespace, data)
}
func createResource(c RESTClient, resource, namespace string, data []byte) error {
return c.Post().Namespace(namespace).Resource(resource).Body(data).Do().Error()
}
func (m *Helper) Update(namespace, name string, overwrite bool, data []byte) error {
c := m.RESTClient
obj, err := m.Codec.Decode(data)
if err != nil {
// We don't know how to handle this object, but update it anyway
return updateResource(c, m.Resource, namespace, name, data)
}
// Attempt to version the object based on client logic.
version, err := m.Versioner.ResourceVersion(obj)
if err != nil {
// We don't know how to version this object, so send it to the server as is
return updateResource(c, m.Resource, namespace, name, data)
}
if version == "" && overwrite {
// Retrieve the current version of the object to overwrite the server object
serverObj, err := c.Get().Namespace(namespace).Resource(m.Resource).Name(name).Do().Get()
if err != nil {
// The object does not exist, but we want it to be created
return updateResource(c, m.Resource, namespace, name, data)
}
serverVersion, err := m.Versioner.ResourceVersion(serverObj)
if err != nil {
return err
}
if err := m.Versioner.SetResourceVersion(obj, serverVersion); err != nil {
return err
}
newData, err := m.Codec.Encode(obj)
if err != nil {
return err
}
data = newData
}
return updateResource(c, m.Resource, namespace, name, data)
}
func updateResource(c RESTClient, resource, namespace, name string, data []byte) error {
return c.Put().Namespace(namespace).Resource(resource).Name(name).Body(data).Do().Error()
}

463
pkg/kubectl/resource/helper_test.go Normal file
View File

@ -0,0 +1,463 @@
/*
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 resource
import (
"bytes"
"errors"
"io"
"io/ioutil"
"net/http"
"reflect"
"strings"
"testing"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/testapi"
"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
)
func objBody(obj runtime.Object) io.ReadCloser {
return ioutil.NopCloser(bytes.NewReader([]byte(runtime.EncodeOrDie(testapi.Codec(), obj))))
}
// splitPath returns the segments for a URL path.
func splitPath(path string) []string {
path = strings.Trim(path, "/")
if path == "" {
return []string{}
}
return strings.Split(path, "/")
}
func TestHelperDelete(t *testing.T) {
tests := []struct {
Err bool
Req func(*http.Request) bool
Resp *http.Response
HttpErr error
}{
{
HttpErr: errors.New("failure"),
Err: true,
},
{
Resp: &http.Response{
StatusCode: http.StatusNotFound,
Body: objBody(&api.Status{Status: api.StatusFailure}),
},
Err: true,
},
{
Resp: &http.Response{
StatusCode: http.StatusOK,
Body: objBody(&api.Status{Status: api.StatusSuccess}),
},
Req: func(req *http.Request) bool {
if req.Method != "DELETE" {
t.Errorf("unexpected method: %#v", req)
return false
}
parts := splitPath(req.URL.Path)
if parts[1] != "bar" {
t.Errorf("url doesn't contain namespace: %#v", req)
return false
}
if parts[2] != "foo" {
t.Errorf("url doesn't contain name: %#v", req)
return false
}
return true
},
},
}
for _, test := range tests {
client := &client.FakeRESTClient{
Codec: testapi.Codec(),
Resp: test.Resp,
Err: test.HttpErr,
}
modifier := &Helper{
RESTClient: client,
}
err := modifier.Delete("bar", "foo")
if (err != nil) != test.Err {
t.Errorf("unexpected error: %t %v", test.Err, err)
}
if err != nil {
continue
}
if test.Req != nil && !test.Req(client.Req) {
t.Errorf("unexpected request: %#v", client.Req)
}
}
}
func TestHelperCreate(t *testing.T) {
expectPost := func(req *http.Request) bool {
if req.Method != "POST" {
t.Errorf("unexpected method: %#v", req)
return false
}
parts := splitPath(req.URL.Path)
if parts[1] != "bar" {
t.Errorf("url doesn't contain namespace: %#v", req)
return false
}
return true
}
tests := []struct {
Resp *http.Response
RespFunc client.HTTPClientFunc
HttpErr error
Modify bool
Object runtime.Object
ExpectObject runtime.Object
Err bool
Req func(*http.Request) bool
}{
{
HttpErr: errors.New("failure"),
Err: true,
},
{
Resp: &http.Response{
StatusCode: http.StatusNotFound,
Body: objBody(&api.Status{Status: api.StatusFailure}),
},
Err: true,
},
{
Resp: &http.Response{
StatusCode: http.StatusOK,
Body: objBody(&api.Status{Status: api.StatusSuccess}),
},
Object: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}},
ExpectObject: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}},
Req: expectPost,
},
{
Modify: false,
Object: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo", ResourceVersion: "10"}},
ExpectObject: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo", ResourceVersion: "10"}},
Resp: &http.Response{StatusCode: http.StatusOK, Body: objBody(&api.Status{Status: api.StatusSuccess})},
Req: expectPost,
},
{
Modify: true,
Object: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo", ResourceVersion: "10"}},
ExpectObject: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}},
Resp: &http.Response{StatusCode: http.StatusOK, Body: objBody(&api.Status{Status: api.StatusSuccess})},
Req: expectPost,
},
}
for i, test := range tests {
client := &client.FakeRESTClient{
Codec: testapi.Codec(),
Resp: test.Resp,
Err: test.HttpErr,
}
if test.RespFunc != nil {
client.Client = test.RespFunc
}
modifier := &Helper{
RESTClient: client,
Codec: testapi.Codec(),
Versioner: testapi.MetadataAccessor(),
}
data := []byte{}
if test.Object != nil {
data = []byte(runtime.EncodeOrDie(testapi.Codec(), test.Object))
}
err := modifier.Create("bar", test.Modify, data)
if (err != nil) != test.Err {
t.Errorf("%d: unexpected error: %t %v", i, test.Err, err)
}
if err != nil {
continue
}
if test.Req != nil && !test.Req(client.Req) {
t.Errorf("%d: unexpected request: %#v", i, client.Req)
}
body, err := ioutil.ReadAll(client.Req.Body)
if err != nil {
t.Fatalf("%d: unexpected error: %#v", i, err)
}
t.Logf("got body: %s", string(body))
expect := []byte{}
if test.ExpectObject != nil {
expect = []byte(runtime.EncodeOrDie(testapi.Codec(), test.ExpectObject))
}
if !reflect.DeepEqual(expect, body) {
t.Errorf("%d: unexpected body: %s", i, string(body))
}
}
}
func TestHelperGet(t *testing.T) {
tests := []struct {
Err bool
Req func(*http.Request) bool
Resp *http.Response
HttpErr error
}{
{
HttpErr: errors.New("failure"),
Err: true,
},
{
Resp: &http.Response{
StatusCode: http.StatusNotFound,
Body: objBody(&api.Status{Status: api.StatusFailure}),
},
Err: true,
},
{
Resp: &http.Response{
StatusCode: http.StatusOK,
Body: objBody(&api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}),
},
Req: func(req *http.Request) bool {
if req.Method != "GET" {
t.Errorf("unexpected method: %#v", req)
return false
}
parts := splitPath(req.URL.Path)
if parts[1] != "bar" {
t.Errorf("url doesn't contain namespace: %#v", req)
return false
}
if parts[2] != "foo" {
t.Errorf("url doesn't contain name: %#v", req)
return false
}
return true
},
},
}
for _, test := range tests {
client := &client.FakeRESTClient{
Codec: testapi.Codec(),
Resp: test.Resp,
Err: test.HttpErr,
}
modifier := &Helper{
RESTClient: client,
}
obj, err := modifier.Get("bar", "foo")
if (err != nil) != test.Err {
t.Errorf("unexpected error: %t %v", test.Err, err)
}
if err != nil {
continue
}
if obj.(*api.Pod).Name != "foo" {
t.Errorf("unexpected object: %#v", obj)
}
if test.Req != nil && !test.Req(client.Req) {
t.Errorf("unexpected request: %#v", client.Req)
}
}
}
func TestHelperList(t *testing.T) {
tests := []struct {
Err bool
Req func(*http.Request) bool
Resp *http.Response
HttpErr error
}{
{
HttpErr: errors.New("failure"),
Err: true,
},
{
Resp: &http.Response{
StatusCode: http.StatusNotFound,
Body: objBody(&api.Status{Status: api.StatusFailure}),
},
Err: true,
},
{
Resp: &http.Response{
StatusCode: http.StatusOK,
Body: objBody(&api.PodList{
Items: []api.Pod{{
ObjectMeta: api.ObjectMeta{Name: "foo"},
},
},
}),
},
Req: func(req *http.Request) bool {
if req.Method != "GET" {
t.Errorf("unexpected method: %#v", req)
return false
}
if req.URL.Path != "/ns/bar" {
t.Errorf("url doesn't contain name: %#v", req.URL)
return false
}
if req.URL.Query().Get("labels") != labels.SelectorFromSet(labels.Set{"foo": "baz"}).String() {
t.Errorf("url doesn't contain query parameters: %#v", req.URL)
return false
}
return true
},
},
}
for _, test := range tests {
client := &client.FakeRESTClient{
Codec: testapi.Codec(),
Resp: test.Resp,
Err: test.HttpErr,
}
modifier := &Helper{
RESTClient: client,
}
obj, err := modifier.List("bar", labels.SelectorFromSet(labels.Set{"foo": "baz"}))
if (err != nil) != test.Err {
t.Errorf("unexpected error: %t %v", test.Err, err)
}
if err != nil {
continue
}
if obj.(*api.PodList).Items[0].Name != "foo" {
t.Errorf("unexpected object: %#v", obj)
}
if test.Req != nil && !test.Req(client.Req) {
t.Errorf("unexpected request: %#v", client.Req)
}
}
}
func TestHelperUpdate(t *testing.T) {
expectPut := func(req *http.Request) bool {
if req.Method != "PUT" {
t.Errorf("unexpected method: %#v", req)
return false
}
parts := splitPath(req.URL.Path)
if parts[1] != "bar" {
t.Errorf("url doesn't contain namespace: %#v", req.URL)
return false
}
if parts[2] != "foo" {
t.Errorf("url doesn't contain name: %#v", req)
return false
}
return true
}
tests := []struct {
Resp *http.Response
RespFunc client.HTTPClientFunc
HttpErr error
Overwrite bool
Object runtime.Object
ExpectObject runtime.Object
Err bool
Req func(*http.Request) bool
}{
{
HttpErr: errors.New("failure"),
Err: true,
},
{
Object: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}},
Resp: &http.Response{
StatusCode: http.StatusNotFound,
Body: objBody(&api.Status{Status: api.StatusFailure}),
},
Err: true,
},
{
Object: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}},
ExpectObject: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}},
Resp: &http.Response{
StatusCode: http.StatusOK,
Body: objBody(&api.Status{Status: api.StatusSuccess}),
},
Req: expectPut,
},
{
Object: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}},
ExpectObject: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo", ResourceVersion: "10"}},
Overwrite: true,
RespFunc: func(req *http.Request) (*http.Response, error) {
if req.Method == "PUT" {
return &http.Response{StatusCode: http.StatusOK, Body: objBody(&api.Status{Status: api.StatusSuccess})}, nil
}
return &http.Response{StatusCode: http.StatusOK, Body: objBody(&api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo", ResourceVersion: "10"}})}, nil
},
Req: expectPut,
},
{
Object: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo", ResourceVersion: "10"}},
ExpectObject: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo", ResourceVersion: "10"}},
Resp: &http.Response{StatusCode: http.StatusOK, Body: objBody(&api.Status{Status: api.StatusSuccess})},
Req: expectPut,
},
}
for i, test := range tests {
client := &client.FakeRESTClient{
Codec: testapi.Codec(),
Resp: test.Resp,
Err: test.HttpErr,
}
if test.RespFunc != nil {
client.Client = test.RespFunc
}
modifier := &Helper{
RESTClient: client,
Codec: testapi.Codec(),
Versioner: testapi.MetadataAccessor(),
}
data := []byte{}
if test.Object != nil {
data = []byte(runtime.EncodeOrDie(testapi.Codec(), test.Object))
}
err := modifier.Update("bar", "foo", test.Overwrite, data)
if (err != nil) != test.Err {
t.Errorf("%d: unexpected error: %t %v", i, test.Err, err)
}
if err != nil {
continue
}
if test.Req != nil && !test.Req(client.Req) {
t.Errorf("%d: unexpected request: %#v", i, client.Req)
}
body, err := ioutil.ReadAll(client.Req.Body)
if err != nil {
t.Fatalf("%d: unexpected error: %#v", i, err)
}
t.Logf("got body: %s", string(body))
expect := []byte{}
if test.ExpectObject != nil {
expect = []byte(runtime.EncodeOrDie(testapi.Codec(), test.ExpectObject))
}
if !reflect.DeepEqual(expect, body) {
t.Errorf("%d: unexpected body: %s", i, string(body))
}
}
}

44
pkg/kubectl/resource/interfaces.go Normal file
View File

@ -0,0 +1,44 @@
/*
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 resource
import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
)
// RESTClient is a client helper for dealing with RESTful resources
// in a generic way.
type RESTClient interface {
Get() *client.Request
Post() *client.Request
Delete() *client.Request
Put() *client.Request
}
// ClientMapper retrieves a client object for a given mapping
type ClientMapper interface {
ClientForMapping(mapping *meta.RESTMapping) (RESTClient, error)
}
// ClientMapperFunc implements ClientMapper for a function
type ClientMapperFunc func(mapping *meta.RESTMapping) (RESTClient, error)
// ClientForMapping implements ClientMapper
func (f ClientMapperFunc) ClientForMapping(mapping *meta.RESTMapping) (RESTClient, error) {
return f(mapping)
}

97
pkg/kubectl/resource/mapper.go Normal file
View File

@ -0,0 +1,97 @@
/*
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 resource
import (
"fmt"
"reflect"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
)
// Mapper is a convenience struct for holding references to the three interfaces
// needed to create Info for arbitrary objects.
type Mapper struct {
runtime.ObjectTyper
meta.RESTMapper
ClientMapper
}
// InfoForData creates an Info object for the given data. An error is returned
// if any of the decoding or client lookup steps fail. Name and namespace will be
// set into Info if the mapping's MetadataAccessor can retrieve them.
func (m *Mapper) InfoForData(data []byte, source string) (*Info, error) {
version, kind, err := m.DataVersionAndKind(data)
if err != nil {
return nil, fmt.Errorf("unable to get type info from %q: %v", source, err)
}
mapping, err := m.RESTMapping(kind, version)
if err != nil {
return nil, fmt.Errorf("unable to recognize %q: %v", source, err)
}
obj, err := mapping.Codec.Decode(data)
if err != nil {
return nil, fmt.Errorf("unable to load %q: %v", source, err)
}
client, err := m.ClientForMapping(mapping)
if err != nil {
return nil, fmt.Errorf("unable to connect to a server to handle %q: %v", mapping.Resource, err)
}
name, _ := mapping.MetadataAccessor.Name(obj)
namespace, _ := mapping.MetadataAccessor.Namespace(obj)
resourceVersion, _ := mapping.MetadataAccessor.ResourceVersion(obj)
return &Info{
Mapping: mapping,
Client: client,
Namespace: namespace,
Name: name,
Object: obj,
ResourceVersion: resourceVersion,
}, nil
}
// InfoForData creates an Info object for the given Object. An error is returned
// if the object cannot be introspected. Name and namespace will be set into Info
// if the mapping's MetadataAccessor can retrieve them.
func (m *Mapper) InfoForObject(obj runtime.Object) (*Info, error) {
version, kind, err := m.ObjectVersionAndKind(obj)
if err != nil {
return nil, fmt.Errorf("unable to get type info from the object %q: %v", reflect.TypeOf(obj), err)
}
mapping, err := m.RESTMapping(kind, version)
if err != nil {
return nil, fmt.Errorf("unable to recognize %q: %v", kind, err)
}
client, err := m.ClientForMapping(mapping)
if err != nil {
return nil, fmt.Errorf("unable to connect to a server to handle %q: %v", mapping.Resource, err)
}
name, _ := mapping.MetadataAccessor.Name(obj)
namespace, _ := mapping.MetadataAccessor.Namespace(obj)
resourceVersion, _ := mapping.MetadataAccessor.ResourceVersion(obj)
return &Info{
Mapping: mapping,
Client: client,
Namespace: namespace,
Name: name,
Object: obj,
ResourceVersion: resourceVersion,
}, nil
}

80
pkg/kubectl/resource/selector.go Normal file
View File

@ -0,0 +1,80 @@
/*
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 resource
import (
"github.com/golang/glog"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
)
// Selector is a Visitor for resources that match a label selector.
type Selector struct {
Client RESTClient
Mapping *meta.RESTMapping
Namespace string
Selector labels.Selector
}
// NewSelector creates a resource selector which hides details of getting items by their label selector.
func NewSelector(client RESTClient, mapping *meta.RESTMapping, namespace string, selector labels.Selector) *Selector {
return &Selector{
Client: client,
Mapping: mapping,
Namespace: namespace,
Selector: selector,
}
}
// Visit implements Visitor
func (r *Selector) Visit(fn VisitorFunc) error {
list, err := NewHelper(r.Client, r.Mapping).List(r.Namespace, r.Selector)
if err != nil {
if errors.IsBadRequest(err) || errors.IsNotFound(err) {
if r.Selector.Empty() {
glog.V(2).Infof("Unable to list %q: %v", r.Mapping.Resource, err)
} else {
glog.V(2).Infof("Unable to find %q that match the selector %q: %v", r.Mapping.Resource, r.Selector, err)
}
return nil
}
return err
}
accessor := r.Mapping.MetadataAccessor
resourceVersion, _ := accessor.ResourceVersion(list)
info := &Info{
Client: r.Client,
Mapping: r.Mapping,
Namespace: r.Namespace,
Object: list,
ResourceVersion: resourceVersion,
}
return fn(info)
}
func (r *Selector) Watch(resourceVersion string) (watch.Interface, error) {
return NewHelper(r.Client, r.Mapping).Watch(r.Namespace, resourceVersion, r.Selector, labels.Everything())
}
// ResourceMapping returns the mapping for this resource and implements ResourceMapping
func (r *Selector) ResourceMapping() *meta.RESTMapping {
return r.Mapping
}

422
pkg/kubectl/resource/visitor.go Normal file
View File

@ -0,0 +1,422 @@
/*
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 resource
import (
"encoding/json"
"fmt"
"io"
"io/ioutil"
"net/http"
"net/url"
"os"
"path/filepath"
"github.com/golang/glog"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/errors"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
)
// Visitor lets clients walk a list of resources.
type Visitor interface {
Visit(VisitorFunc) error
}
// VisitorFunc implements the Visitor interface for a matching function
type VisitorFunc func(*Info) error
// Watchable describes a resource that can be watched for changes that occur on the server,
// beginning after the provided resource version.
type Watchable interface {
Watch(resourceVersion string) (watch.Interface, error)
}
// ResourceMapping allows an object to return the resource mapping associated with
// the resource or resources it represents.
type ResourceMapping interface {
ResourceMapping() *meta.RESTMapping
}
// Info contains temporary info to execute a REST call, or show the results
// of an already completed REST call.
type Info struct {
Client RESTClient
Mapping *meta.RESTMapping
Namespace string
Name string
// Optional, this is the most recent value returned by the server if available
runtime.Object
// Optional, this is the most recent resource version the server knows about for
// this type of resource. It may not match the resource version of the object,
// but if set it should be equal to or newer than the resource version of the
// object (however the server defines resource version).
ResourceVersion string
}
// NewInfo returns a new info object
func NewInfo(client RESTClient, mapping *meta.RESTMapping, namespace, name string) *Info {
return &Info{
Client: client,
Mapping: mapping,
Namespace: namespace,
Name: name,
}
}
// Visit implements Visitor
func (i *Info) Visit(fn VisitorFunc) error {
return fn(i)
}
func (i *Info) Get() error {
obj, err := NewHelper(i.Client, i.Mapping).Get(i.Namespace, i.Name)
if err != nil {
return err
}
i.Object = obj
i.ResourceVersion, _ = i.Mapping.MetadataAccessor.ResourceVersion(obj)
return nil
}
// Watch returns server changes to this object after it was retrieved.
func (i *Info) Watch(resourceVersion string) (watch.Interface, error) {
return NewHelper(i.Client, i.Mapping).WatchSingle(i.Namespace, i.Name, resourceVersion)
}
// ResourceMapping returns the mapping for this resource and implements ResourceMapping
func (i *Info) ResourceMapping() *meta.RESTMapping {
return i.Mapping
}
// VisitorList implements Visit for the sub visitors it contains. The first error
// returned from a child Visitor will terminate iteration.
type VisitorList []Visitor
// Visit implements Visitor
func (l VisitorList) Visit(fn VisitorFunc) error {
for i := range l {
if err := l[i].Visit(fn); err != nil {
return err
}
}
return nil
}
// EagerVisitorList implements Visit for the sub visitors it contains. All errors
// will be captured and returned at the end of iteration.
type EagerVisitorList []Visitor
// Visit implements Visitor, and gathers errors that occur during processing until
// all sub visitors have been visited.
func (l EagerVisitorList) Visit(fn VisitorFunc) error {
errs := []error(nil)
for i := range l {
if err := l[i].Visit(func(info *Info) error {
if err := fn(info); err != nil {
errs = append(errs, err)
}
return nil
}); err != nil {
errs = append(errs, err)
}
}
return errors.NewAggregate(errs)
}
// PathVisitor visits a given path and returns an object representing the file
// at that path.
type PathVisitor struct {
*Mapper
// The file path to load
Path string
// Whether to ignore files that are not recognized as API objects
IgnoreErrors bool
}
func (v *PathVisitor) Visit(fn VisitorFunc) error {
data, err := ioutil.ReadFile(v.Path)
if err != nil {
return fmt.Errorf("unable to read %q: %v", v.Path, err)
}
info, err := v.Mapper.InfoForData(data, v.Path)
if err != nil {
if v.IgnoreErrors {
return err
}
glog.V(2).Infof("Unable to load file %q: %v", v.Path, err)
return nil
}
return fn(info)
}
// DirectoryVisitor loads the specified files from a directory and passes them
// to visitors.
type DirectoryVisitor struct {
*Mapper
// The directory or file to start from
Path string
// Whether directories are recursed
Recursive bool
// The file extensions to include. If empty, all files are read.
Extensions []string
// Whether to ignore files that are not recognized as API objects
IgnoreErrors bool
}
func (v *DirectoryVisitor) ignoreFile(path string) bool {
if len(v.Extensions) == 0 {
return false
}
ext := filepath.Ext(path)
for _, s := range v.Extensions {
if s == ext {
return false
}
}
return true
}
func (v *DirectoryVisitor) Visit(fn VisitorFunc) error {
return filepath.Walk(v.Path, func(path string, fi os.FileInfo, err error) error {
if err != nil {
return err
}
if fi.IsDir() {
if path != v.Path && !v.Recursive {
return filepath.SkipDir
}
return nil
}
if v.ignoreFile(path) {
return nil
}
data, err := ioutil.ReadFile(path)
if err != nil {
return fmt.Errorf("unable to read %q: %v", path, err)
}
info, err := v.Mapper.InfoForData(data, path)
if err != nil {
if v.IgnoreErrors {
return err
}
glog.V(2).Infof("Unable to load file %q: %v", path, err)
return nil
}
return fn(info)
})
}
// URLVisitor downloads the contents of a URL, and if successful, returns
// an info object representing the downloaded object.
type URLVisitor struct {
*Mapper
URL *url.URL
}
func (v *URLVisitor) Visit(fn VisitorFunc) error {
res, err := http.Get(v.URL.String())
if err != nil {
return fmt.Errorf("unable to access URL %q: %v\n", v.URL, err)
}
defer res.Body.Close()
if res.StatusCode != 200 {
return fmt.Errorf("unable to read URL %q, server reported %d %s", v.URL, res.StatusCode, res.Status)
}
data, err := ioutil.ReadAll(res.Body)
if err != nil {
return fmt.Errorf("unable to read URL %q: %v\n", v.URL, err)
}
info, err := v.Mapper.InfoForData(data, v.URL.String())
if err != nil {
return err
}
return fn(info)
}
// DecoratedVisitor will invoke the decorators in order prior to invoking the visitor function
// passed to Visit. An error will terminate the visit.
type DecoratedVisitor struct {
visitor Visitor
decorators []VisitorFunc
}
// NewDecoratedVisitor will create a visitor that invokes the provided visitor functions before
// the user supplied visitor function is invoked, giving them the opportunity to mutate the Info
// object or terminate early with an error.
func NewDecoratedVisitor(v Visitor, fn ...VisitorFunc) Visitor {
if len(fn) == 0 {
return v
}
return DecoratedVisitor{v, fn}
}
// Visit implements Visitor
func (v DecoratedVisitor) Visit(fn VisitorFunc) error {
return v.visitor.Visit(func(info *Info) error {
for i := range v.decorators {
if err := v.decorators[i](info); err != nil {
return err
}
}
return fn(info)
})
}
// FlattenListVisitor flattens any objects that runtime.ExtractList recognizes as a list
// - has an "Items" public field that is a slice of runtime.Objects or objects satisfying
// that interface - into multiple Infos. An error on any sub item (for instance, if a List
// contains an object that does not have a registered client or resource) will terminate
// the visit.
// TODO: allow errors to be aggregated?
type FlattenListVisitor struct {
Visitor
*Mapper
}
// NewFlattenListVisitor creates a visitor that will expand list style runtime.Objects
// into individual items and then visit them individually.
func NewFlattenListVisitor(v Visitor, mapper *Mapper) Visitor {
return FlattenListVisitor{v, mapper}
}
func (v FlattenListVisitor) Visit(fn VisitorFunc) error {
return v.Visitor.Visit(func(info *Info) error {
if info.Object == nil {
return fn(info)
}
items, err := runtime.ExtractList(info.Object)
if err != nil {
return fn(info)
}
for i := range items {
item, err := v.InfoForObject(items[i])
if err != nil {
return err
}
if len(info.ResourceVersion) != 0 {
item.ResourceVersion = info.ResourceVersion
}
if err := fn(item); err != nil {
return err
}
}
return nil
})
}
// StreamVisitor reads objects from an io.Reader and walks them. A stream visitor can only be
// visited once.
// TODO: depends on objects being in JSON format before being passed to decode - need to implement
// a stream decoder method on runtime.Codec to properly handle this.
type StreamVisitor struct {
io.Reader
*Mapper
Source string
IgnoreErrors bool
}
// NewStreamVisitor creates a visitor that will return resources that were encoded into the provided
// stream. If ignoreErrors is set, unrecognized or invalid objects will be skipped and logged.
func NewStreamVisitor(r io.Reader, mapper *Mapper, source string, ignoreErrors bool) Visitor {
return &StreamVisitor{r, mapper, source, ignoreErrors}
}
// Visit implements Visitor over a stream.
func (v *StreamVisitor) Visit(fn VisitorFunc) error {
d := json.NewDecoder(v.Reader)
for {
ext := runtime.RawExtension{}
if err := d.Decode(&ext); err != nil {
if err == io.EOF {
return nil
}
return err
}
info, err := v.InfoForData(ext.RawJSON, v.Source)
if err != nil {
if v.IgnoreErrors {
glog.V(2).Infof("Unable to read item from stream %q: %v", err)
glog.V(4).Infof("Unreadable: %s", string(ext.RawJSON))
continue
}
return err
}
if err := fn(info); err != nil {
return err
}
}
return nil
}
func UpdateObjectNamespace(info *Info) error {
if info.Object != nil {
return info.Mapping.MetadataAccessor.SetNamespace(info.Object, info.Namespace)
}
return nil
}
// SetNamespace ensures that every Info object visited will have a namespace
// set. If info.Object is set, it will be mutated as well.
func SetNamespace(namespace string) VisitorFunc {
return func(info *Info) error {
if len(info.Namespace) == 0 {
info.Namespace = namespace
UpdateObjectNamespace(info)
}
return nil
}
}
// RequireNamespace will either set a namespace if none is provided on the
// Info object, or if the namespace is set and does not match the provided
// value, returns an error. This is intended to guard against administrators
// accidentally operating on resources outside their namespace.
func RequireNamespace(namespace string) VisitorFunc {
return func(info *Info) error {
if len(info.Namespace) == 0 {
info.Namespace = namespace
UpdateObjectNamespace(info)
return nil
}
if info.Namespace != namespace {
return fmt.Errorf("the namespace from the provided object %q does not match the namespace %q. You must pass '--namespace=%s' to perform this operation.", info.Namespace, namespace, info.Namespace)
}
return nil
}
}
// RetrieveLatest updates the Object on each Info by invoking a standard client
// Get.
func RetrieveLatest(info *Info) error {
if len(info.Name) == 0 || len(info.Namespace) == 0 {
return nil
}
obj, err := NewHelper(info.Client, info.Mapping).Get(info.Namespace, info.Name)
if err != nil {
return err
}
info.Object = obj
info.ResourceVersion, _ = info.Mapping.MetadataAccessor.ResourceVersion(obj)
return nil
}