Merge pull request #3152 from smarterclayton/resource_args_builder

cmd/kubectl/kubectl.go

@@ -19,11 +19,15 @@ package main
 import (
 	"os"
 
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/client/clientcmd"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubectl/cmd"
+
+	"github.com/golang/glog"
 )
 
 func main() {
-	clientBuilder := clientcmd.NewInteractiveClientConfig(clientcmd.Config{}, "", &clientcmd.ConfigOverrides{}, os.Stdin)
-	cmd.NewFactory(clientBuilder).Run(os.Stdout)
+	cmd := cmd.NewFactory().NewKubectlCommand(os.Stdout)
+	if err := cmd.Execute(); err != nil {
+		glog.Errorf("error: %v", err)
+		os.Exit(1)
+	}
 }

pkg/kubectl/cmd/cmd.go

@@ -33,6 +33,7 @@ import (
 
 	"github.com/golang/glog"
 	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
 )
 
 const (
@@ -42,55 +43,108 @@ const (
 // Factory provides abstractions that allow the Kubectl command to be extended across multiple types
 // of resources and different API sets.
 type Factory struct {
-	ClientConfig clientcmd.ClientConfig
-	Mapper       meta.RESTMapper
-	Typer        runtime.ObjectTyper
-	Client       func(cmd *cobra.Command, mapping *meta.RESTMapping) (kubectl.RESTClient, error)
-	Describer    func(cmd *cobra.Command, mapping *meta.RESTMapping) (kubectl.Describer, error)
-	Printer      func(cmd *cobra.Command, mapping *meta.RESTMapping, noHeaders bool) (kubectl.ResourcePrinter, error)
-	Validator    func(*cobra.Command) (validation.Schema, error)
+	clients *clientCache
+	flags   *pflag.FlagSet
+
+	Mapper meta.RESTMapper
+	Typer  runtime.ObjectTyper
+
+	// Returns a client for accessing Kubernetes resources or an error.
+	Client func(cmd *cobra.Command) (*client.Client, error)
+	// Returns a client.Config for accessing the Kubernetes server.
+	ClientConfig func(cmd *cobra.Command) (*client.Config, error)
+	// Returns a RESTClient for working with the specified RESTMapping or an error. This is intended
+	// for working with arbitrary resources and is not guaranteed to point to a Kubernetes APIServer.
+	RESTClient func(cmd *cobra.Command, mapping *meta.RESTMapping) (kubectl.RESTClient, error)
+	// Returns a Describer for displaying the specified RESTMapping type or an error.
+	Describer func(cmd *cobra.Command, mapping *meta.RESTMapping) (kubectl.Describer, error)
+	// Returns a Printer for formatting objects of the given type or an error.
+	Printer func(cmd *cobra.Command, mapping *meta.RESTMapping, noHeaders bool) (kubectl.ResourcePrinter, error)
+	// Returns a schema that can validate objects stored on disk.
+	Validator func(*cobra.Command) (validation.Schema, error)
 }
 
 // NewFactory creates a factory with the default Kubernetes resources defined
-func NewFactory(clientConfig clientcmd.ClientConfig) *Factory {
-	ret := &Factory{
-		ClientConfig: clientConfig,
-		Mapper:       latest.RESTMapper,
-		Typer:        api.Scheme,
-		Printer: func(cmd *cobra.Command, mapping *meta.RESTMapping, noHeaders bool) (kubectl.ResourcePrinter, error) {
-			return kubectl.NewHumanReadablePrinter(noHeaders), nil
-		},
+func NewFactory() *Factory {
+	mapper := kubectl.ShortcutExpander{latest.RESTMapper}
+
+	flags := pflag.NewFlagSet("", pflag.ContinueOnError)
+	clientConfig := DefaultClientConfig(flags)
+	clients := &clientCache{
+		clients: make(map[string]*client.Client),
+		loader:  clientConfig,
 	}
 
-	ret.Validator = func(cmd *cobra.Command) (validation.Schema, error) {
-		if GetFlagBool(cmd, "validate") {
-			client, err := getClient(ret.ClientConfig, GetFlagBool(cmd, FlagMatchBinaryVersion))
+	return &Factory{
+		clients: clients,
+		flags:   flags,
+
+		Mapper: mapper,
+		Typer:  api.Scheme,
+
+		Client: func(cmd *cobra.Command) (*client.Client, error) {
+			return clients.ClientForVersion("")
+		},
+		ClientConfig: func(cmd *cobra.Command) (*client.Config, error) {
+			return clients.ClientConfigForVersion("")
+		},
+		RESTClient: func(cmd *cobra.Command, mapping *meta.RESTMapping) (kubectl.RESTClient, error) {
+			client, err := clients.ClientForVersion(mapping.APIVersion)
 			if err != nil {
 				return nil, err
 			}
-			return &clientSwaggerSchema{client, api.Scheme}, nil
-		} else {
+			return client.RESTClient, nil
+		},
+		Describer: func(cmd *cobra.Command, mapping *meta.RESTMapping) (kubectl.Describer, error) {
+			client, err := clients.ClientForVersion(mapping.APIVersion)
+			if err != nil {
+				return nil, err
+			}
+			describer, ok := kubectl.DescriberFor(mapping.Kind, client)
+			if !ok {
+				return nil, fmt.Errorf("no description has been implemented for %q", mapping.Kind)
+			}
+			return describer, nil
+		},
+		Printer: func(cmd *cobra.Command, mapping *meta.RESTMapping, noHeaders bool) (kubectl.ResourcePrinter, error) {
+			return kubectl.NewHumanReadablePrinter(noHeaders), nil
+		},
+		Validator: func(cmd *cobra.Command) (validation.Schema, error) {
+			if GetFlagBool(cmd, "validate") {
+				client, err := clients.ClientForVersion("")
+				if err != nil {
+					return nil, err
+				}
+				return &clientSwaggerSchema{client, api.Scheme}, nil
+			}
 			return validation.NullSchema{}, nil
-		}
-	}
-	ret.Client = func(cmd *cobra.Command, mapping *meta.RESTMapping) (kubectl.RESTClient, error) {
-		return getClient(ret.ClientConfig, GetFlagBool(cmd, FlagMatchBinaryVersion))
+		},
 	}
-	ret.Describer = func(cmd *cobra.Command, mapping *meta.RESTMapping) (kubectl.Describer, error) {
-		client, err := getClient(ret.ClientConfig, GetFlagBool(cmd, FlagMatchBinaryVersion))
-		if err != nil {
-			return nil, err
-		}
-		describer, ok := kubectl.DescriberFor(mapping.Kind, client)
-		if !ok {
-			return nil, fmt.Errorf("no description has been implemented for %q", mapping.Kind)
-		}
-		return describer, nil
+}
+
+// BindFlags adds any flags that are common to all kubectl sub commands.
+func (f *Factory) BindFlags(flags *pflag.FlagSet) {
+	// any flags defined by external projects (not part of pflags)
+	util.AddAllFlagsToPFlagSet(flags)
+
+	if f.flags != nil {
+		f.flags.VisitAll(func(flag *pflag.Flag) {
+			flags.AddFlag(flag)
+		})
 	}
-	return ret
+
+	// Globally persistent flags across all subcommands.
+	// TODO Change flag names to consts to allow safer lookup from subcommands.
+	// TODO Add a verbose flag that turns on glog logging. Probably need a way
+	// to do that automatically for every subcommand.
+	flags.BoolVar(&f.clients.matchVersion, FlagMatchBinaryVersion, false, "Require server version to match client version")
+	flags.String("ns-path", os.Getenv("HOME")+"/.kubernetes_ns", "Path to the namespace info file that holds the namespace context to use for CLI requests.")
+	flags.StringP("namespace", "n", "", "If present, the namespace scope for this CLI request.")
+	flags.Bool("validate", false, "If true, use a schema to validate the input before sending it")
 }
 
-func (f *Factory) Run(out io.Writer) {
+// NewKubectlCommand creates the `kubectl` command and its nested children.
+func (f *Factory) NewKubectlCommand(out io.Writer) *cobra.Command {
 	// Parent command to which all subcommands are added.
 	cmds := &cobra.Command{
 		Use:   "kubectl",
@@ -101,15 +155,7 @@ Find more information at https://github.com/GoogleCloudPlatform/kubernetes.`,
 		Run: runHelp,
 	}
 
-	util.AddAllFlagsToPFlagSet(cmds.PersistentFlags())
-	f.ClientConfig = getClientConfig(cmds)
-
-	// Globally persistent flags across all subcommands.
-	// TODO Change flag names to consts to allow safer lookup from subcommands.
-	cmds.PersistentFlags().Bool(FlagMatchBinaryVersion, false, "Require server version to match client version")
-	cmds.PersistentFlags().String("ns-path", os.Getenv("HOME")+"/.kubernetes_ns", "Path to the namespace info file that holds the namespace context to use for CLI requests.")
-	cmds.PersistentFlags().StringP("namespace", "n", "", "If present, the namespace scope for this CLI request.")
-	cmds.PersistentFlags().Bool("validate", false, "If true, use a schema to validate the input before sending it")
+	f.BindFlags(cmds.PersistentFlags())
 
 	cmds.AddCommand(f.NewCmdVersion(out))
 	cmds.AddCommand(f.NewCmdProxy(out))
@@ -125,12 +171,10 @@ Find more information at https://github.com/GoogleCloudPlatform/kubernetes.`,
 	cmds.AddCommand(f.NewCmdLog(out))
 	cmds.AddCommand(f.NewCmdRollingUpdate(out))
 
-	if err := cmds.Execute(); err != nil {
-		os.Exit(1)
-	}
+	return cmds
 }
 
-// getClientBuilder creates a clientcmd.ClientConfig that has a hierarchy like this:
+// DefaultClientConfig creates a clientcmd.ClientConfig with the following hierarchy:
 //   1.  Use the kubeconfig builder.  The number of merges and overrides here gets a little crazy.  Stay with me.
 //       1.  Merge together the kubeconfig itself.  This is done with the following hierarchy and merge rules:
 //           1.  CommandLineLocation - this parsed from the command line, so it must be late bound
@@ -162,13 +206,13 @@ Find more information at https://github.com/GoogleCloudPlatform/kubernetes.`,
 //           2.  If the command line does not specify one, and the auth info has conflicting techniques, fail.
 //           3.  If the command line specifies one and the auth info specifies another, honor the command line technique.
 //   2.  Use default values and potentially prompt for auth information
-func getClientConfig(cmd *cobra.Command) clientcmd.ClientConfig {
+func DefaultClientConfig(flags *pflag.FlagSet) clientcmd.ClientConfig {
 	loadingRules := clientcmd.NewClientConfigLoadingRules()
 	loadingRules.EnvVarPath = os.Getenv(clientcmd.RecommendedConfigPathEnvVar)
-	cmd.PersistentFlags().StringVar(&loadingRules.CommandLinePath, "kubeconfig", "", "Path to the kubeconfig file to use for CLI requests.")
+	flags.StringVar(&loadingRules.CommandLinePath, "kubeconfig", "", "Path to the kubeconfig file to use for CLI requests.")
 
 	overrides := &clientcmd.ConfigOverrides{}
-	overrides.BindFlags(cmd.PersistentFlags(), clientcmd.RecommendedConfigOverrideFlags(""))
+	overrides.BindFlags(flags, clientcmd.RecommendedConfigOverrideFlags(""))
 	clientConfig := clientcmd.NewInteractiveDeferredLoadingClientConfig(loadingRules, overrides, os.Stdin)
 
 	return clientConfig
@@ -246,18 +290,55 @@ func (c *clientSwaggerSchema) ValidateBytes(data []byte) error {
 	return schema.ValidateBytes(data)
 }
 
-// TODO Need to only run server version match once per client host creation
-func getClient(clientConfig clientcmd.ClientConfig, matchServerVersion bool) (*client.Client, error) {
-	config, err := clientConfig.ClientConfig()
-	if err != nil {
-		return nil, err
-	}
+// clientCache caches previously loaded clients for reuse, and ensures MatchServerVersion
+// is invoked only once
+type clientCache struct {
+	loader        clientcmd.ClientConfig
+	clients       map[string]*client.Client
+	defaultConfig *client.Config
+	matchVersion  bool
+}
 
-	if matchServerVersion {
-		err := client.MatchesServerVersion(config)
+// ClientConfigForVersion returns the correct config for a server
+func (c *clientCache) ClientConfigForVersion(version string) (*client.Config, error) {
+	if c.defaultConfig == nil {
+		config, err := c.loader.ClientConfig()
 		if err != nil {
 			return nil, err
 		}
+		c.defaultConfig = config
+
+		if c.matchVersion {
+			if err := client.MatchesServerVersion(config); err != nil {
+				return nil, err
+			}
+		}
+	}
+
+	// TODO: remove when SetKubernetesDefaults gets added
+	if len(version) == 0 {
+		version = c.defaultConfig.Version
+	}
+
+	// TODO: have a better config copy method
+	config := *c.defaultConfig
+
+	// TODO: call new client.SetKubernetesDefaults method
+	// instead of doing this
+	config.Version = version
+	return &config, nil
+}
+
+// ClientForVersion initializes or reuses a client for the specified version, or returns an
+// error if that is not possible
+func (c *clientCache) ClientForVersion(version string) (*client.Client, error) {
+	config, err := c.ClientConfigForVersion(version)
+	if err != nil {
+		return nil, err
+	}
+
+	if client, ok := c.clients[config.Version]; ok {
+		return client, nil
 	}
 
 	client, err := client.New(config)
@@ -265,5 +346,6 @@ func getClient(clientConfig clientcmd.ClientConfig, matchServerVersion bool) (*c
 		return nil, err
 	}
 
+	c.clients[config.Version] = client
 	return client, nil
 }

pkg/kubectl/cmd/cmd_test.go

@@ -100,7 +100,7 @@ func NewTestFactory() (*Factory, *testFactory, runtime.Codec) {
 	return &Factory{
 		Mapper: mapper,
 		Typer:  scheme,
-		Client: func(*cobra.Command, *meta.RESTMapping) (kubectl.RESTClient, error) {
+		RESTClient: func(*cobra.Command, *meta.RESTMapping) (kubectl.RESTClient, error) {
 			return t.Client, t.Err
 		},
 		Describer: func(*cobra.Command, *meta.RESTMapping) (kubectl.Describer, error) {

pkg/kubectl/cmd/create.go

@@ -20,7 +20,7 @@ import (
 	"fmt"
 	"io"
 
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubectl"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubectl/resource"
 	"github.com/spf13/cobra"
 )
 
@@ -46,7 +46,7 @@ Examples:
 			schema, err := f.Validator(cmd)
 			checkErr(err)
 			mapping, namespace, name, data := ResourceFromFile(cmd, filename, f.Typer, f.Mapper, schema)
-			client, err := f.Client(cmd, mapping)
+			client, err := f.RESTClient(cmd, mapping)
 			checkErr(err)
 
 			// use the default namespace if not specified, or check for conflict with the file's namespace
@@ -57,7 +57,7 @@ Examples:
 				checkErr(err)
 			}
 
-			err = kubectl.NewRESTHelper(client, mapping).Create(namespace, true, data)
+			err = resource.NewHelper(client, mapping).Create(namespace, true, data)
 			checkErr(err)
 			fmt.Fprintf(out, "%s\n", name)
 		},

pkg/kubectl/cmd/createall.go

@@ -79,7 +79,7 @@ Examples:
   <creates all resources listed in config.json>`,
 		Run: func(cmd *cobra.Command, args []string) {
 			clientFunc := func(mapper *meta.RESTMapping) (config.RESTClientPoster, error) {
-				client, err := f.Client(cmd, mapper)
+				client, err := f.RESTClient(cmd, mapper)
 				checkErr(err)
 				return client, nil
 			}

pkg/kubectl/cmd/delete.go

@@ -23,7 +23,7 @@ import (
 	"github.com/golang/glog"
 	"github.com/spf13/cobra"
 
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubectl"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubectl/resource"
 )
 
 func (f *Factory) NewCmdDelete(out io.Writer) *cobra.Command {
@@ -59,9 +59,9 @@ Examples:
 			checkErr(err)
 			selector := GetFlagString(cmd, "selector")
 			found := 0
-			ResourcesFromArgsOrFile(cmd, args, filename, selector, f.Typer, f.Mapper, f.Client, schema).Visit(func(r *ResourceInfo) error {
+			ResourcesFromArgsOrFile(cmd, args, filename, selector, f.Typer, f.Mapper, f.RESTClient, schema).Visit(func(r *resource.Info) error {
 				found++
-				if err := kubectl.NewRESTHelper(r.Client, r.Mapping).Delete(r.Namespace, r.Name); err != nil {
+				if err := resource.NewHelper(r.Client, r.Mapping).Delete(r.Namespace, r.Name); err != nil {
 					return err
 				}
 				fmt.Fprintf(out, "%s\n", r.Name)

pkg/kubectl/cmd/get.go

@@ -21,7 +21,9 @@ import (
 	"io"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubectl"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubectl/resource"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 
 	"github.com/spf13/cobra"
 )
@@ -54,7 +56,7 @@ Examples:
 			labelSelector, err := labels.ParseSelector(selector)
 			checkErr(err)
 
-			client, err := f.Client(cmd, mapping)
+			client, err := f.RESTClient(cmd, mapping)
 			checkErr(err)
 
 			outputFormat := GetFlagString(cmd, "output")
@@ -70,8 +72,13 @@ Examples:
 			printer, err := kubectl.GetPrinter(outputFormat, templateFile, outputVersion, mapping.ObjectConvertor, defaultPrinter)
 			checkErr(err)
 
-			restHelper := kubectl.NewRESTHelper(client, mapping)
-			obj, err := restHelper.Get(namespace, name, labelSelector)
+			restHelper := resource.NewHelper(client, mapping)
+			var obj runtime.Object
+			if len(name) == 0 {
+				obj, err = restHelper.List(namespace, labelSelector)
+			} else {
+				obj, err = restHelper.Get(namespace, name)
+			}
 			checkErr(err)
 
 			isWatch, isWatchOnly := GetFlagBool(cmd, "watch"), GetFlagBool(cmd, "watch-only")

pkg/kubectl/cmd/helpers.go

@@ -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 {

pkg/kubectl/cmd/log.go

@@ -21,8 +21,6 @@ import (
 	"strconv"
 
 	"github.com/spf13/cobra"
-
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
 )
 
 func (f *Factory) NewCmdLog(out io.Writer) *cobra.Command {
@@ -46,9 +44,7 @@ Examples:
 			}
 
 			namespace := GetKubeNamespace(cmd)
-			config, err := f.ClientConfig.ClientConfig()
-			checkErr(err)
-			client, err := client.New(config)
+			client, err := f.Client(cmd)
 			checkErr(err)
 
 			podID := args[0]

pkg/kubectl/cmd/proxy.go

@@ -33,7 +33,7 @@ func (f *Factory) NewCmdProxy(out io.Writer) *cobra.Command {
 			port := GetFlagInt(cmd, "port")
 			glog.Infof("Starting to serve on localhost:%d", port)
 
-			clientConfig, err := f.ClientConfig.ClientConfig()
+			clientConfig, err := f.ClientConfig(cmd)
 			checkErr(err)
 
 			server, err := kubectl.NewProxyServer(GetFlagString(cmd, "www"), clientConfig, port)

pkg/kubectl/cmd/resource.go

@@ -18,122 +18,19 @@ package cmd
 
 import (
 	"fmt"
-	"log"
 	"strings"
 
-	"github.com/golang/glog"
 	"github.com/spf13/cobra"
 
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/validation"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubectl"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubectl/resource"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 )
 
-// ResourceInfo contains temporary info to execute REST call
-type ResourceInfo struct {
-	Client    kubectl.RESTClient
-	Mapping   *meta.RESTMapping
-	Namespace string
-	Name      string
-
-	// Optional, this is the most recent value returned by the server if available
-	runtime.Object
-}
-
-// ResourceVisitor lets clients walk the list of resources
-type ResourceVisitor interface {
-	Visit(func(*ResourceInfo) error) error
-}
-
-type ResourceVisitorList []ResourceVisitor
-
-// Visit implements ResourceVisitor
-func (l ResourceVisitorList) Visit(fn func(r *ResourceInfo) error) error {
-	for i := range l {
-		if err := l[i].Visit(fn); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func NewResourceInfo(client kubectl.RESTClient, mapping *meta.RESTMapping, namespace, name string) *ResourceInfo {
-	return &ResourceInfo{
-		Client:    client,
-		Mapping:   mapping,
-		Namespace: namespace,
-		Name:      name,
-	}
-}
-
-// Visit implements ResourceVisitor
-func (r *ResourceInfo) Visit(fn func(r *ResourceInfo) error) error {
-	return fn(r)
-}
-
-// ResourceSelector is a facade for all the resources fetched via label selector
-type ResourceSelector struct {
-	Client    kubectl.RESTClient
-	Mapping   *meta.RESTMapping
-	Namespace string
-	Selector  labels.Selector
-}
-
-// NewResourceSelector creates a resource selector which hides details of getting items by their label selector.
-func NewResourceSelector(client kubectl.RESTClient, mapping *meta.RESTMapping, namespace string, selector labels.Selector) *ResourceSelector {
-	return &ResourceSelector{
-		Client:    client,
-		Mapping:   mapping,
-		Namespace: namespace,
-		Selector:  selector,
-	}
-}
-
-// Visit implements ResourceVisitor
-func (r *ResourceSelector) Visit(fn func(r *ResourceInfo) error) error {
-	list, err := kubectl.NewRESTHelper(r.Client, r.Mapping).List(r.Namespace, r.Selector)
-	if err != nil {
-		if errors.IsBadRequest(err) || errors.IsNotFound(err) {
-			glog.V(2).Infof("Unable to perform a label selector query on %s with labels %s: %v", r.Mapping.Resource, r.Selector, err)
-			return nil
-		}
-		return err
-	}
-	items, err := runtime.ExtractList(list)
-	if err != nil {
-		return err
-	}
-	accessor := meta.NewAccessor()
-	for i := range items {
-		name, err := accessor.Name(items[i])
-		if err != nil {
-			// items without names cannot be visited
-			glog.V(2).Infof("Found %s with labels %s, but can't access the item by name.", r.Mapping.Resource, r.Selector)
-			continue
-		}
-		item := &ResourceInfo{
-			Client:    r.Client,
-			Mapping:   r.Mapping,
-			Namespace: r.Namespace,
-			Name:      name,
-			Object:    items[i],
-		}
-		if err := fn(item); err != nil {
-			if errors.IsNotFound(err) {
-				glog.V(2).Infof("Found %s named %q, but can't be accessed now: %v", r.Mapping.Resource, name, err)
-				return nil
-			}
-			log.Printf("got error for resource %s: %v", r.Mapping.Resource, err)
-			return err
-		}
-	}
-	return nil
-}
-
 // ResourcesFromArgsOrFile computes a list of Resources by extracting info from filename or args. It will
 // handle label selectors provided.
 func ResourcesFromArgsOrFile(
@@ -144,7 +41,7 @@ func ResourcesFromArgsOrFile(
 	mapper meta.RESTMapper,
 	clientBuilder func(cmd *cobra.Command, mapping *meta.RESTMapping) (kubectl.RESTClient, error),
 	schema validation.Schema,
-) ResourceVisitor {
+) resource.Visitor {
 
 	// handling filename & resource id
 	if len(selector) == 0 {
@@ -152,34 +49,34 @@ func ResourcesFromArgsOrFile(
 		client, err := clientBuilder(cmd, mapping)
 		checkErr(err)
 
-		return NewResourceInfo(client, mapping, namespace, name)
+		return resource.NewInfo(client, mapping, namespace, name)
 	}
 
 	labelSelector, err := labels.ParseSelector(selector)
 	checkErr(err)
 
 	namespace := GetKubeNamespace(cmd)
-	visitors := ResourceVisitorList{}
+	visitors := resource.VisitorList{}
 
 	if len(args) != 1 {
 		usageError(cmd, "Must specify the type of resource")
 	}
 	types := SplitResourceArgument(args[0])
 	for _, arg := range types {
-		resource := kubectl.ExpandResourceShortcut(arg)
-		if len(resource) == 0 {
-			usageError(cmd, "Unknown resource %s", resource)
+		resourceName := arg
+		if len(resourceName) == 0 {
+			usageError(cmd, "Unknown resource %s", resourceName)
 		}
-		version, kind, err := mapper.VersionAndKindForResource(resource)
+		version, kind, err := mapper.VersionAndKindForResource(resourceName)
 		checkErr(err)
 
-		mapping, err := mapper.RESTMapping(version, kind)
+		mapping, err := mapper.RESTMapping(kind, version)
 		checkErr(err)
 
 		client, err := clientBuilder(cmd, mapping)
 		checkErr(err)
 
-		visitors = append(visitors, NewResourceSelector(client, mapping, namespace, labelSelector))
+		visitors = append(visitors, resource.NewSelector(client, mapping, namespace, labelSelector))
 	}
 	return visitors
 }
@@ -194,7 +91,7 @@ func ResourceFromArgsOrFile(cmd *cobra.Command, args []string, filename string,
 	}
 
 	if len(args) == 2 {
-		resource := kubectl.ExpandResourceShortcut(args[0])
+		resource := args[0]
 		namespace = GetKubeNamespace(cmd)
 		name = args[1]
 		if len(name) == 0 || len(resource) == 0 {
@@ -232,7 +129,7 @@ func ResourceFromArgs(cmd *cobra.Command, args []string, mapper meta.RESTMapper)
 		usageError(cmd, "Must provide resource and name command line params")
 	}
 
-	resource := kubectl.ExpandResourceShortcut(args[0])
+	resource := args[0]
 	namespace = GetKubeNamespace(cmd)
 	name = args[1]
 	if len(name) == 0 || len(resource) == 0 {
@@ -255,7 +152,7 @@ func ResourceOrTypeFromArgs(cmd *cobra.Command, args []string, mapper meta.RESTM
 		usageError(cmd, "Must provide resource or a resource and name as command line params")
 	}
 
-	resource := kubectl.ExpandResourceShortcut(args[0])
+	resource := args[0]
 	if len(resource) == 0 {
 		usageError(cmd, "Must provide resource or a resource and name as command line params")
 	}

pkg/kubectl/cmd/rollingupdate.go

@@ -21,7 +21,6 @@ import (
 	"io"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubectl"
 	"github.com/spf13/cobra"
 )
@@ -69,9 +68,7 @@ $ cat frontend-v2.json | kubectl rollingupdate frontend-v1 -f -
 			err = CompareNamespaceFromFile(cmd, namespace)
 			checkErr(err)
 
-			config, err := f.ClientConfig.ClientConfig()
-			checkErr(err)
-			client, err := client.New(config)
+			client, err := f.Client(cmd)
 			checkErr(err)
 
 			obj, err := mapping.Codec.Decode(data)

pkg/kubectl/cmd/update.go

@@ -20,7 +20,7 @@ import (
 	"fmt"
 	"io"
 
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubectl"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubectl/resource"
 	"github.com/spf13/cobra"
 )
 
@@ -46,13 +46,13 @@ Examples:
 			schema, err := f.Validator(cmd)
 			checkErr(err)
 			mapping, namespace, name, data := ResourceFromFile(cmd, filename, f.Typer, f.Mapper, schema)
-			client, err := f.Client(cmd, mapping)
+			client, err := f.RESTClient(cmd, mapping)
 			checkErr(err)
 
 			err = CompareNamespaceFromFile(cmd, namespace)
 			checkErr(err)
 
-			err = kubectl.NewRESTHelper(client, mapping).Update(namespace, name, true, data)
+			err = resource.NewHelper(client, mapping).Update(namespace, name, true, data)
 			checkErr(err)
 			fmt.Fprintf(out, "%s\n", name)
 		},

pkg/kubectl/cmd/version.go

@@ -21,7 +21,6 @@ import (
 
 	"github.com/spf13/cobra"
 
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubectl"
 )
 
@@ -32,14 +31,13 @@ func (f *Factory) NewCmdVersion(out io.Writer) *cobra.Command {
 		Run: func(cmd *cobra.Command, args []string) {
 			if GetFlagBool(cmd, "client") {
 				kubectl.GetClientVersion(out)
-			} else {
-				config, err := f.ClientConfig.ClientConfig()
-				checkErr(err)
-				client, err := client.New(config)
-				checkErr(err)
-
-				kubectl.GetVersion(out, client)
+				return
 			}
+
+			client, err := f.Client(cmd)
+			checkErr(err)
+
+			kubectl.GetVersion(out, client)
 		},
 	}
 	cmd.Flags().BoolP("client", "c", false, "Client version only (no server required)")

pkg/kubectl/kubectl.go

@@ -26,6 +26,7 @@ import (
 	"strings"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
@@ -111,12 +112,23 @@ func makeImageList(spec *api.PodSpec) string {
 	return strings.Join(listOfImages(spec), ",")
 }
 
-// ExpandResourceShortcut will return the expanded version of resource
+// ShortcutExpander is a RESTMapper that can be used for Kubernetes
+// resources.
+type ShortcutExpander struct {
+	meta.RESTMapper
+}
+
+// VersionAndKindForResource implements meta.RESTMapper. It expands the resource first, then invokes the wrapped
+// mapper.
+func (e ShortcutExpander) VersionAndKindForResource(resource string) (defaultVersion, kind string, err error) {
+	resource = expandResourceShortcut(resource)
+	return e.RESTMapper.VersionAndKindForResource(resource)
+}
+
+// expandResourceShortcut will return the expanded version of resource
 // (something that a pkg/api/meta.RESTMapper can understand), if it is
 // indeed a shortcut. Otherwise, will return resource unmodified.
-// TODO: Combine with RESTMapper stuff to provide a general solution
-// to this problem.
-func ExpandResourceShortcut(resource string) string {
+func expandResourceShortcut(resource string) string {
 	shortForms := map[string]string{
 		"po": "pods",
 		"rc": "replicationcontrollers",

pkg/kubectl/resource/builder.go

+/*
+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()
+}

pkg/kubectl/resource/builder_test.go

+/*
+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)
+	}
+}

pkg/kubectl/resource/doc.go

+/*
+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

pkg/kubectl/resthelper.go → pkg/kubectl/resource/helper.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package kubectl
+package resource
 
 import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
@@ -23,9 +23,10 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
 )
 
-// RESTHelper provides methods for retrieving or mutating a RESTful
+// Helper provides methods for retrieving or mutating a RESTful
 // resource.
-type RESTHelper struct {
+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
@@ -36,9 +37,9 @@ type RESTHelper struct {
 	Versioner runtime.ResourceVersioner
 }
 
-// NewRESTHelper creates a RESTHelper from a ResourceMapping
-func NewRESTHelper(client RESTClient, mapping *meta.RESTMapping) *RESTHelper {
-	return &RESTHelper{
+// 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,
@@ -46,15 +47,25 @@ func NewRESTHelper(client RESTClient, mapping *meta.RESTMapping) *RESTHelper {
 	}
 }
 
-func (m *RESTHelper) Get(namespace, name string, selector labels.Selector) (runtime.Object, error) {
-	return m.RESTClient.Get().Resource(m.Resource).Namespace(namespace).Name(name).SelectorParam("labels", selector).Do().Get()
+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 *RESTHelper) List(namespace string, selector labels.Selector) (runtime.Object, error) {
-	return m.RESTClient.Get().Resource(m.Resource).Namespace(namespace).SelectorParam("labels", selector).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 *RESTHelper) Watch(namespace, resourceVersion string, labelSelector, fieldSelector labels.Selector) (watch.Interface, error) {
+func (m *Helper) Watch(namespace, resourceVersion string, labelSelector, fieldSelector labels.Selector) (watch.Interface, error) {
 	return m.RESTClient.Get().
 		Prefix("watch").
 		Namespace(namespace).
@@ -65,11 +76,26 @@ func (m *RESTHelper) Watch(namespace, resourceVersion string, labelSelector, fie
 		Watch()
 }
 
-func (m *RESTHelper) Delete(namespace, name string) error {
-	return m.RESTClient.Delete().Namespace(namespace).Resource(m.Resource).Name(name).Do().Error()
+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 *RESTHelper) Create(namespace string, modify bool, data []byte) error {
+func (m *Helper) Create(namespace string, modify bool, data []byte) error {
 	if modify {
 		obj, err := m.Codec.Decode(data)
 		if err != nil {
@@ -102,7 +128,7 @@ func createResource(c RESTClient, resource, namespace string, data []byte) error
 	return c.Post().Namespace(namespace).Resource(resource).Body(data).Do().Error()
 }
 
-func (m *RESTHelper) Update(namespace, name string, overwrite bool, data []byte) error {
+func (m *Helper) Update(namespace, name string, overwrite bool, data []byte) error {
 	c := m.RESTClient
 
 	obj, err := m.Codec.Decode(data)
@@ -119,7 +145,7 @@ func (m *RESTHelper) Update(namespace, name string, overwrite bool, data []byte)
 	}
 	if version == "" && overwrite {
 		// Retrieve the current version of the object to overwrite the server object
-		serverObj, err := c.Get().Resource(m.Resource).Name(name).Do().Get()
+		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)

pkg/kubectl/resthelper_test.go → pkg/kubectl/resource/helper_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package kubectl
+package resource
 
 import (
 	"bytes"
@@ -46,7 +46,7 @@ func splitPath(path string) []string {
 	return strings.Split(path, "/")
 }
 
-func TestRESTHelperDelete(t *testing.T) {
+func TestHelperDelete(t *testing.T) {
 	tests := []struct {
 		Err     bool
 		Req     func(*http.Request) bool
@@ -93,7 +93,7 @@ func TestRESTHelperDelete(t *testing.T) {
 			Resp:  test.Resp,
 			Err:   test.HttpErr,
 		}
-		modifier := &RESTHelper{
+		modifier := &Helper{
 			RESTClient: client,
 		}
 		err := modifier.Delete("bar", "foo")
@@ -109,7 +109,7 @@ func TestRESTHelperDelete(t *testing.T) {
 	}
 }
 
-func TestRESTHelperCreate(t *testing.T) {
+func TestHelperCreate(t *testing.T) {
 	expectPost := func(req *http.Request) bool {
 		if req.Method != "POST" {
 			t.Errorf("unexpected method: %#v", req)
@@ -178,7 +178,7 @@ func TestRESTHelperCreate(t *testing.T) {
 		if test.RespFunc != nil {
 			client.Client = test.RespFunc
 		}
-		modifier := &RESTHelper{
+		modifier := &Helper{
 			RESTClient: client,
 			Codec:      testapi.Codec(),
 			Versioner:  testapi.MetadataAccessor(),
@@ -213,7 +213,7 @@ func TestRESTHelperCreate(t *testing.T) {
 	}
 }
 
-func TestRESTHelperGet(t *testing.T) {
+func TestHelperGet(t *testing.T) {
 	tests := []struct {
 		Err     bool
 		Req     func(*http.Request) bool
@@ -260,10 +260,10 @@ func TestRESTHelperGet(t *testing.T) {
 			Resp:  test.Resp,
 			Err:   test.HttpErr,
 		}
-		modifier := &RESTHelper{
+		modifier := &Helper{
 			RESTClient: client,
 		}
-		obj, err := modifier.Get("bar", "foo", labels.Everything())
+		obj, err := modifier.Get("bar", "foo")
 		if (err != nil) != test.Err {
 			t.Errorf("unexpected error: %t %v", test.Err, err)
 		}
@@ -279,7 +279,77 @@ func TestRESTHelperGet(t *testing.T) {
 	}
 }
 
-func TestRESTHelperUpdate(t *testing.T) {
+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)
@@ -358,7 +428,7 @@ func TestRESTHelperUpdate(t *testing.T) {
 		if test.RespFunc != nil {
 			client.Client = test.RespFunc
 		}
-		modifier := &RESTHelper{
+		modifier := &Helper{
 			RESTClient: client,
 			Codec:      testapi.Codec(),
 			Versioner:  testapi.MetadataAccessor(),

pkg/kubectl/resource/interfaces.go

+/*
+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)
+}

pkg/kubectl/resource/mapper.go

+/*
+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
+}

pkg/kubectl/resource/selector.go

+/*
+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
+}

pkg/kubectl/resource/visitor.go

+/*
+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
+}