Fix validation by moving it into the resource builder.

Also always print an error for unknown field.

pkg/kubectl/resource/builder.go

@@ -24,6 +24,7 @@ import (
 	"strings"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/validation"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
@@ -62,6 +63,8 @@ type Builder struct {
 
 	singleResourceType bool
 	continueOnError    bool
+
+	schema validation.Schema
 }
 
 type resourceTuple struct {
@@ -77,6 +80,11 @@ func NewBuilder(mapper meta.RESTMapper, typer runtime.ObjectTyper, clientMapper
 	}
 }
 
+func (b *Builder) Schema(schema validation.Schema) *Builder {
+	b.schema = schema
+	return b
+}
+
 // 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)).
@@ -105,6 +113,7 @@ func (b *Builder) URL(urls ...*url.URL) *Builder {
 		b.paths = append(b.paths, &URLVisitor{
 			Mapper: b.mapper,
 			URL:    u,
+			Schema: b.schema,
 		})
 	}
 	return b
@@ -123,7 +132,7 @@ func (b *Builder) Stdin() *Builder {
 // 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))
+	b.paths = append(b.paths, NewStreamVisitor(r, b.mapper, b.schema, name, b.continueOnError))
 	return b
 }
 
@@ -150,12 +159,14 @@ func (b *Builder) Path(paths ...string) *Builder {
 				Extensions:   []string{".json", ".yaml", ".yml"},
 				Recursive:    false,
 				IgnoreErrors: b.continueOnError,
+				Schema:       b.schema,
 			}
 		} else {
 			visitor = &PathVisitor{
 				Mapper:       b.mapper,
 				Path:         p,
 				IgnoreErrors: b.continueOnError,
+				Schema:       b.schema,
 			}
 		}
 		b.paths = append(b.paths, visitor)

pkg/kubectl/resource/visitor.go

@@ -29,6 +29,7 @@ import (
 	"github.com/golang/glog"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/validation"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/errors"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/yaml"
@@ -180,6 +181,20 @@ func (l EagerVisitorList) Visit(fn VisitorFunc) error {
 	return errors.NewAggregate(errs)
 }
 
+func ValidateSchema(data []byte, schema validation.Schema) error {
+	if schema == nil {
+		return nil
+	}
+	data, err := yaml.ToJSON(data)
+	if err != nil {
+		return fmt.Errorf("error converting to YAML: %v", err)
+	}
+	if err := schema.ValidateBytes(data); err != nil {
+		return fmt.Errorf("error validating data: %v", err)
+	}
+	return nil
+}
+
 // PathVisitor visits a given path and returns an object representing the file
 // at that path.
 type PathVisitor struct {
@@ -188,6 +203,8 @@ type PathVisitor struct {
 	Path string
 	// Whether to ignore files that are not recognized as API objects
 	IgnoreErrors bool
+	// Schema for validation
+	Schema validation.Schema
 }
 
 func (v *PathVisitor) Visit(fn VisitorFunc) error {
@@ -195,6 +212,9 @@ func (v *PathVisitor) Visit(fn VisitorFunc) error {
 	if err != nil {
 		return fmt.Errorf("unable to read %q: %v", v.Path, err)
 	}
+	if err := ValidateSchema(data, v.Schema); err != nil {
+		return err
+	}
 	info, err := v.Mapper.InfoForData(data, v.Path)
 	if err != nil {
 		if v.IgnoreErrors {
@@ -218,6 +238,8 @@ type DirectoryVisitor struct {
 	Extensions []string
 	// Whether to ignore files that are not recognized as API objects
 	IgnoreErrors bool
+	// Schema for validation
+	Schema validation.Schema
 }
 
 func (v *DirectoryVisitor) ignoreFile(path string) bool {
@@ -253,6 +275,9 @@ func (v *DirectoryVisitor) Visit(fn VisitorFunc) error {
 		if err != nil {
 			return fmt.Errorf("unable to read %q: %v", path, err)
 		}
+		if err := ValidateSchema(data, v.Schema); err != nil {
+			return err
+		}
 		info, err := v.Mapper.InfoForData(data, path)
 		if err != nil {
 			if v.IgnoreErrors {
@@ -269,7 +294,8 @@ func (v *DirectoryVisitor) Visit(fn VisitorFunc) error {
 // an info object representing the downloaded object.
 type URLVisitor struct {
 	*Mapper
-	URL *url.URL
+	URL    *url.URL
+	Schema validation.Schema
 }
 
 func (v *URLVisitor) Visit(fn VisitorFunc) error {
@@ -285,6 +311,9 @@ func (v *URLVisitor) Visit(fn VisitorFunc) error {
 	if err != nil {
 		return fmt.Errorf("unable to read URL %q: %v\n", v.URL, err)
 	}
+	if err := ValidateSchema(data, v.Schema); err != nil {
+		return err
+	}
 	info, err := v.Mapper.InfoForData(data, v.URL.String())
 	if err != nil {
 		return err
@@ -379,6 +408,7 @@ type StreamVisitor struct {
 
 	Source       string
 	IgnoreErrors bool
+	Schema       validation.Schema
 }
 
 // NewStreamVisitor creates a visitor that will return resources that were encoded into the provided
@@ -386,8 +416,8 @@ type StreamVisitor struct {
 // empty stream is treated as an error for now.
 // TODO: convert ignoreErrors into a func(data, error, count) bool that consumers can use to decide
 // what to do with ignored errors.
-func NewStreamVisitor(r io.Reader, mapper *Mapper, source string, ignoreErrors bool) Visitor {
-	return &StreamVisitor{r, mapper, source, ignoreErrors}
+func NewStreamVisitor(r io.Reader, mapper *Mapper, schema validation.Schema, source string, ignoreErrors bool) Visitor {
+	return &StreamVisitor{r, mapper, source, ignoreErrors, schema}
 }
 
 // Visit implements Visitor over a stream.
@@ -405,6 +435,9 @@ func (v *StreamVisitor) Visit(fn VisitorFunc) error {
 		if len(ext.RawJSON) == 0 || bytes.Equal(ext.RawJSON, []byte("null")) {
 			continue
 		}
+		if err := ValidateSchema(ext.RawJSON, v.Schema); err != nil {
+			return err
+		}
 		info, err := v.InfoForData(ext.RawJSON, v.Source)
 		if err != nil {
 			if v.IgnoreErrors {