func validateVolumeMounts(mounts []VolumeMount, volumes util.StringSet) errs.ErrorList {
	allErrs := errs.ErrorList{}

	for i := range mounts {
		mnt := &mounts[i] // so we can set default values
		if len(mnt.Name) == 0 {
			allErrs = append(allErrs, errs.NewInvalid("VolumeMount.Name", mnt.Name))
		} else if !volumes.Has(mnt.Name) {
			allErrs = append(allErrs, errs.NewNotFound("VolumeMount.Name", mnt.Name))
		}
		if len(mnt.MountPath) == 0 {
			// Backwards compat.
			if len(mnt.Path) == 0 {
				allErrs = append(allErrs, errs.NewInvalid("VolumeMount.MountPath", mnt.MountPath))
			} else {
				glog.Warning("DEPRECATED: VolumeMount.Path has been replaced by VolumeMount.MountPath")
				mnt.MountPath = mnt.Path
				mnt.Path = ""
			}
		}
		if len(mnt.MountType) != 0 {
			glog.Warning("DEPRECATED: VolumeMount.MountType will be removed. The Volume struct will handle types")
		}
	}
	return allErrs
}

func validateContainers(containers []Container, volumes util.StringSet) errs.ErrorList {
	allErrs := errs.ErrorList{}

	allNames := util.StringSet{}
	for i := range containers {
		ctr := &containers[i] // so we can set default values
		if !util.IsDNSLabel(ctr.Name) {
			allErrs = append(allErrs, errs.NewInvalid("Container.Name", ctr.Name))
		} else if allNames.Has(ctr.Name) {
			allErrs = append(allErrs, errs.NewDuplicate("Container.Name", ctr.Name))
		} else {
			allNames.Insert(ctr.Name)
		}
		if len(ctr.Image) == 0 {
			allErrs = append(allErrs, errs.NewInvalid("Container.Image", ctr.Name))
		}
		allErrs = append(allErrs, validatePorts(ctr.Ports)...)
		allErrs = append(allErrs, validateEnv(ctr.Env)...)
		allErrs = append(allErrs, validateVolumeMounts(ctr.VolumeMounts, volumes)...)
	}
	// Check for colliding ports across all containers.
	// TODO(thockin): This really is dependent on the network config of the host (IP per pod?)
	// and the config of the new manifest.  But we have not specced that out yet, so we'll just
	// make some assumptions for now.  As of now, pods share a network namespace, which means that
	// every Port.HostPort across the whole pod must be unique.
	allErrs = append(allErrs, checkHostPortConflicts(containers)...)

	return allErrs
}

func TestCheckInvalidErr(t *testing.T) {
	tests := []struct {
		err      error
		expected string
	}{
		{
			errors.NewInvalid("Invalid1", "invalidation", fielderrors.ValidationErrorList{fielderrors.NewFieldInvalid("Cause", "single", "details")}),
			`Error from server: Invalid1 "invalidation" is invalid: Cause: invalid value 'single', Details: details`,
		},
		{
			errors.NewInvalid("Invalid2", "invalidation", fielderrors.ValidationErrorList{fielderrors.NewFieldInvalid("Cause", "multi1", "details"), fielderrors.NewFieldInvalid("Cause", "multi2", "details")}),
			`Error from server: Invalid2 "invalidation" is invalid: [Cause: invalid value 'multi1', Details: details, Cause: invalid value 'multi2', Details: details]`,
		},
		{
			errors.NewInvalid("Invalid3", "invalidation", fielderrors.ValidationErrorList{}),
			`Error from server: Invalid3 "invalidation" is invalid: <nil>`,
		},
	}

	var errReturned string
	errHandle := func(err string) {
		errReturned = err
	}

	for _, test := range tests {
		checkErr(test.err, errHandle)

		if errReturned != test.expected {
			t.Fatalf("Got: %s, expected: %s", errReturned, test.expected)
		}
	}
}

func validatePorts(ports []Port) errs.ErrorList {
	allErrs := errs.ErrorList{}

	allNames := util.StringSet{}
	for i := range ports {
		port := &ports[i] // so we can set default values
		if len(port.Name) > 0 {
			if len(port.Name) > 63 || !util.IsDNSLabel(port.Name) {
				allErrs.Append(errs.NewInvalid("Port.Name", port.Name))
			} else if allNames.Has(port.Name) {
				allErrs.Append(errs.NewDuplicate("Port.name", port.Name))
			} else {
				allNames.Insert(port.Name)
			}
		}
		if !util.IsValidPortNum(port.ContainerPort) {
			allErrs.Append(errs.NewInvalid("Port.ContainerPort", port.ContainerPort))
		}
		if port.HostPort == 0 {
			port.HostPort = port.ContainerPort
		} else if !util.IsValidPortNum(port.HostPort) {
			allErrs.Append(errs.NewInvalid("Port.HostPort", port.HostPort))
		}
		if len(port.Protocol) == 0 {
			port.Protocol = "TCP"
		} else if !supportedPortProtocols.Has(strings.ToUpper(port.Protocol)) {
			allErrs.Append(errs.NewNotSupported("Port.Protocol", port.Protocol))
		}
	}
	return allErrs
}

func (rs *REST) Update(ctx api.Context, obj runtime.Object) (<-chan apiserver.RESTResult, error) {
	service := obj.(*api.Service)
	if !api.ValidNamespace(ctx, &service.ObjectMeta) {
		return nil, errors.NewConflict("service", service.Namespace, fmt.Errorf("Service.Namespace does not match the provided context"))
	}
	if errs := validation.ValidateService(service, rs.registry, ctx); len(errs) > 0 {
		return nil, errors.NewInvalid("service", service.Name, errs)
	}
	return apiserver.MakeAsync(func() (runtime.Object, error) {
		cur, err := rs.registry.GetService(ctx, service.Name)
		if err != nil {
			return nil, err
		}
		if service.Spec.PortalIP != cur.Spec.PortalIP {
			// TODO: Would be nice to pass "field is immutable" to users.
			el := errors.ValidationErrorList{errors.NewFieldInvalid("spec.portalIP", service.Spec.PortalIP)}
			return nil, errors.NewInvalid("service", service.Name, el)
		}
		// Copy over non-user fields.
		service.Spec.ProxyPort = cur.Spec.ProxyPort
		// TODO: check to see if external load balancer status changed
		err = rs.registry.UpdateService(ctx, service)
		if err != nil {
			return nil, err
		}
		return rs.registry.GetService(ctx, service.Name)
	}), nil
}

func validatePorts(ports []Port) errs.ErrorList {
	allErrs := errs.ErrorList{}

	allNames := util.StringSet{}
	for i := range ports {
		pErrs := errs.ErrorList{}
		port := &ports[i] // so we can set default values
		if len(port.Name) > 0 {
			if len(port.Name) > 63 || !util.IsDNSLabel(port.Name) {
				pErrs = append(pErrs, errs.NewInvalid("name", port.Name))
			} else if allNames.Has(port.Name) {
				pErrs = append(pErrs, errs.NewDuplicate("name", port.Name))
			} else {
				allNames.Insert(port.Name)
			}
		}
		if port.ContainerPort == 0 {
			pErrs = append(pErrs, errs.NewRequired("containerPort", port.ContainerPort))
		} else if !util.IsValidPortNum(port.ContainerPort) {
			pErrs = append(pErrs, errs.NewInvalid("containerPort", port.ContainerPort))
		}
		if port.HostPort != 0 && !util.IsValidPortNum(port.HostPort) {
			pErrs = append(pErrs, errs.NewInvalid("hostPort", port.HostPort))
		}
		if len(port.Protocol) == 0 {
			port.Protocol = "TCP"
		} else if !supportedPortProtocols.Has(strings.ToUpper(port.Protocol)) {
			pErrs = append(pErrs, errs.NewNotSupported("protocol", port.Protocol))
		}
		allErrs = append(allErrs, pErrs.PrefixIndex(i)...)
	}
	return allErrs
}

// ValidateService tests if required fields in the service are set.
func ValidateService(service *Service) errs.ErrorList {
	allErrs := errs.ErrorList{}
	if service.ID == "" {
		allErrs = append(allErrs, errs.NewInvalid("Service.ID", service.ID))
	} else if !util.IsDNS952Label(service.ID) {
		allErrs = append(allErrs, errs.NewInvalid("Service.ID", service.ID))
	}
	if labels.Set(service.Selector).AsSelector().Empty() {
		allErrs = append(allErrs, errs.NewInvalid("Service.Selector", service.Selector))
	}
	return allErrs
}

// Create ensures a pod is bound to a specific host.
func (r *BindingREST) Create(ctx api.Context, obj runtime.Object) (out runtime.Object, err error) {
	binding := obj.(*api.Binding)
	// TODO: move me to a binding strategy
	if len(binding.Target.Kind) != 0 && (binding.Target.Kind != "Node" && binding.Target.Kind != "Minion") {
		return nil, errors.NewInvalid("binding", binding.Name, fielderrors.ValidationErrorList{fielderrors.NewFieldInvalid("to.kind", binding.Target.Kind, "must be empty, 'Node', or 'Minion'")})
	}
	if len(binding.Target.Name) == 0 {
		return nil, errors.NewInvalid("binding", binding.Name, fielderrors.ValidationErrorList{fielderrors.NewFieldRequired("to.name")})
	}
	err = r.assignPod(ctx, binding.Name, binding.Target.Name, binding.Annotations)
	out = &api.Status{Status: api.StatusSuccess}
	return
}

// ValidateReplicationController tests if required fields in the replication controller are set.
func ValidateReplicationController(controller *ReplicationController) errs.ErrorList {
	allErrs := errs.ErrorList{}
	if controller.ID == "" {
		allErrs = append(allErrs, errs.NewInvalid("ReplicationController.ID", controller.ID))
	}
	if labels.Set(controller.DesiredState.ReplicaSelector).AsSelector().Empty() {
		allErrs = append(allErrs, errs.NewInvalid("ReplicationController.ReplicaSelector", controller.DesiredState.ReplicaSelector))
	}
	if controller.DesiredState.Replicas < 0 {
		allErrs = append(allErrs, errs.NewInvalid("ReplicationController.Replicas", controller.DesiredState.Replicas))
	}
	allErrs = append(allErrs, ValidateManifest(&controller.DesiredState.PodTemplate.DesiredState.Manifest)...)
	return allErrs
}

func validateEnv(vars []EnvVar) errs.ErrorList {
	allErrs := errs.ErrorList{}

	for i := range vars {
		ev := &vars[i] // so we can set default values
		if len(ev.Name) == 0 {
			allErrs = append(allErrs, errs.NewInvalid("EnvVar.Name", ev.Name))
		}
		if !util.IsCIdentifier(ev.Name) {
			allErrs = append(allErrs, errs.NewInvalid("EnvVar.Name", ev.Name))
		}
	}
	return allErrs
}

// ValidateReplicationController tests if required fields in the replication controller are set.
func ValidateReplicationController(controller *ReplicationController) []error {
	el := []error{}
	if controller.ID == "" {
		el = append(el, errs.NewInvalid("ReplicationController.ID", controller.ID))
	}
	if labels.Set(controller.DesiredState.ReplicaSelector).AsSelector().Empty() {
		el = append(el, errs.NewInvalid("ReplicationController.ReplicaSelector", controller.DesiredState.ReplicaSelector))
	}
	if controller.DesiredState.Replicas < 0 {
		el = append(el, errs.NewInvalid("ReplicationController.Replicas", controller.DesiredState.Replicas))
	}
	el = append(el, ValidateManifest(&controller.DesiredState.PodTemplate.DesiredState.Manifest)...)
	return el
}

// ValidateService tests if required fields in the service are set.
func ValidateService(service *Service) errs.ErrorList {
	allErrs := errs.ErrorList{}
	if len(service.ID) == 0 {
		allErrs = append(allErrs, errs.NewRequired("id", service.ID))
	} else if !util.IsDNS952Label(service.ID) {
		allErrs = append(allErrs, errs.NewInvalid("id", service.ID))
	}
	if !util.IsValidPortNum(service.Port) {
		allErrs = append(allErrs, errs.NewInvalid("Service.Port", service.Port))
	}
	if labels.Set(service.Selector).AsSelector().Empty() {
		allErrs = append(allErrs, errs.NewRequired("selector", service.Selector))
	}
	return allErrs
}

func validateVolumes(volumes []Volume) (util.StringSet, errs.ErrorList) {
	allErrs := errs.ErrorList{}

	allNames := util.StringSet{}
	for i := range volumes {
		vol := &volumes[i] // so we can set default values
		el := errs.ErrorList{}
		// TODO(thockin) enforce that a source is set once we deprecate the implied form.
		if vol.Source != nil {
			el = validateSource(vol.Source).Prefix("source")
		}
		if len(vol.Name) == 0 {
			el = append(el, errs.NewRequired("name", vol.Name))
		} else if !util.IsDNSLabel(vol.Name) {
			el = append(el, errs.NewInvalid("name", vol.Name))
		} else if allNames.Has(vol.Name) {
			el = append(el, errs.NewDuplicate("name", vol.Name))
		}
		if len(el) == 0 {
			allNames.Insert(vol.Name)
		} else {
			allErrs = append(allErrs, el.PrefixIndex(i)...)
		}
	}
	return allNames, allErrs
}

// Update satisfies the RESTStorage interface.
func (rs *REST) Update(ctx api.Context, obj runtime.Object) (runtime.Object, bool, error) {
	minion, ok := obj.(*api.Node)
	if !ok {
		return nil, false, fmt.Errorf("not a minion: %#v", obj)
	}
	// This is hacky, but minions don't really have a namespace, but kubectl currently automatically
	// stuffs one in there.  Fix it here temporarily until we fix kubectl
	if minion.Namespace == api.NamespaceDefault {
		minion.Namespace = api.NamespaceNone
	}
	// Clear out the self link, if specified, since it's not in the registry either.
	minion.SelfLink = ""

	oldMinion, err := rs.registry.GetMinion(ctx, minion.Name)
	if err != nil {
		return nil, false, err
	}

	if errs := validation.ValidateMinionUpdate(oldMinion, minion); len(errs) > 0 {
		return nil, false, kerrors.NewInvalid("minion", minion.Name, errs)
	}

	if err := rs.registry.UpdateMinion(ctx, minion); err != nil {
		return nil, false, err
	}
	out, err := rs.registry.GetMinion(ctx, minion.Name)
	return out, false, err
}

// Create registers the given ReplicationController.
func (rs *REST) Create(ctx api.Context, obj runtime.Object) (<-chan apiserver.RESTResult, error) {
	controller, ok := obj.(*api.ReplicationController)
	if !ok {
		return nil, fmt.Errorf("not a replication controller: %#v", obj)
	}
	if !api.ValidNamespace(ctx, &controller.ObjectMeta) {
		return nil, errors.NewConflict("controller", controller.Namespace, fmt.Errorf("Controller.Namespace does not match the provided context"))
	}

	if len(controller.Name) == 0 {
		controller.Name = util.NewUUID().String()
	}
	// Pod Manifest ID should be assigned by the pod API
	controller.DesiredState.PodTemplate.DesiredState.Manifest.ID = ""
	if errs := validation.ValidateReplicationController(controller); len(errs) > 0 {
		return nil, errors.NewInvalid("replicationController", controller.Name, errs)
	}

	controller.CreationTimestamp = util.Now()

	return apiserver.MakeAsync(func() (runtime.Object, error) {
		err := rs.registry.CreateController(ctx, controller)
		if err != nil {
			return nil, err
		}
		return rs.registry.GetController(ctx, controller.Name)
	}), nil
}

// Update satisfies the RESTStorage interface.
func (rs *REST) Update(ctx api.Context, obj runtime.Object) (<-chan apiserver.RESTResult, error) {
	minion, ok := obj.(*api.Node)
	if !ok {
		return nil, fmt.Errorf("not a minion: %#v", obj)
	}
	// This is hacky, but minions don't really have a namespace, but kubectl currently automatically
	// stuffs one in there.  Fix it here temporarily until we fix kubectl
	if minion.Namespace == api.NamespaceDefault {
		minion.Namespace = api.NamespaceNone
	}
	// Clear out the self link, if specified, since it's not in the registry either.
	minion.SelfLink = ""

	// TODO: GetMinion will health check the minion, but we shouldn't require the minion to be
	// running for updating labels.
	oldMinion, err := rs.registry.GetMinion(ctx, minion.Name)
	if err != nil {
		return nil, err
	}
	if errs := validation.ValidateMinionUpdate(oldMinion, minion); len(errs) > 0 {
		return nil, kerrors.NewInvalid("minion", minion.Name, errs)
	}

	return apiserver.MakeAsync(func() (runtime.Object, error) {
		err := rs.registry.UpdateMinion(ctx, minion)
		if err != nil {
			return nil, err
		}
		return rs.registry.GetMinion(ctx, minion.Name)
	}), nil
}

// Update replaces a given Route instance with an existing instance in rs.registry.
func (rs *REST) Update(ctx kapi.Context, obj runtime.Object) (runtime.Object, bool, error) {
	route, ok := obj.(*api.Route)
	if !ok {
		return nil, false, errors.NewBadRequest(fmt.Sprintf("not a route: %#v", obj))
	}
	if !kapi.ValidNamespace(ctx, &route.ObjectMeta) {
		return nil, false, errors.NewConflict("route", route.Namespace, fmt.Errorf("Route.Namespace does not match the provided context"))
	}

	old, err := rs.Get(ctx, route.Name)
	if err != nil {
		return nil, false, err
	}
	if errs := validation.ValidateRouteUpdate(route, old.(*api.Route)); len(errs) > 0 {
		return nil, false, errors.NewInvalid("route", route.Name, errs)
	}

	// TODO: Convert to generic etcd
	// TODO: Call ValidateRouteUpdate->ValidateObjectMetaUpdate
	// TODO: In the UpdateStrategy.PrepareForUpdate, set the HostGeneratedAnnotationKey annotation to "false" if the updated route object modifies the host

	err = rs.registry.UpdateRoute(ctx, route)
	if err != nil {
		return nil, false, err
	}
	out, err := rs.registry.GetRoute(ctx, route.Name)
	return out, false, err
}

// Create registers the given ReplicationController.
func (rs *REST) Create(ctx api.Context, obj runtime.Object) (<-chan apiserver.RESTResult, error) {
	controller, ok := obj.(*api.ReplicationController)
	if !ok {
		return nil, fmt.Errorf("not a replication controller: %#v", obj)
	}
	if !api.ValidNamespace(ctx, &controller.ObjectMeta) {
		return nil, errors.NewConflict("controller", controller.Namespace, fmt.Errorf("Controller.Namespace does not match the provided context"))
	}

	if len(controller.Name) == 0 {
		controller.Name = util.NewUUID().String()
	}
	if errs := validation.ValidateReplicationController(controller); len(errs) > 0 {
		return nil, errors.NewInvalid("replicationController", controller.Name, errs)
	}

	api.FillObjectMetaSystemFields(ctx, &controller.ObjectMeta)

	return apiserver.MakeAsync(func() (runtime.Object, error) {
		err := rs.registry.CreateController(ctx, controller)
		if err != nil {
			return nil, err
		}
		return rs.registry.GetController(ctx, controller.Name)
	}), nil
}

// Create a Secret object
func (rs *REST) Create(ctx api.Context, obj runtime.Object) (runtime.Object, error) {
	secret, ok := obj.(*api.Secret)
	if !ok {
		return nil, fmt.Errorf("invalid object type")
	}

	if !api.ValidNamespace(ctx, &secret.ObjectMeta) {
		return nil, errors.NewConflict("secret", secret.Namespace, fmt.Errorf("Secret.Namespace does not match the provided context"))
	}

	if len(secret.Name) == 0 {
		secret.Name = string(util.NewUUID())
	}

	if errs := validation.ValidateSecret(secret); len(errs) > 0 {
		return nil, errors.NewInvalid("secret", secret.Name, errs)
	}
	api.FillObjectMetaSystemFields(ctx, &secret.ObjectMeta)

	err := rs.registry.CreateWithName(ctx, secret.Name, secret)
	if err != nil {
		return nil, err
	}
	return rs.registry.Get(ctx, secret.Name)
}

// Update updates a LimitRange object.
func (rs *REST) Update(ctx api.Context, obj runtime.Object) (runtime.Object, bool, error) {
	limitRange, ok := obj.(*api.LimitRange)
	if !ok {
		return nil, false, fmt.Errorf("invalid object type")
	}

	if !api.ValidNamespace(ctx, &limitRange.ObjectMeta) {
		return nil, false, errors.NewConflict("limitRange", limitRange.Namespace, fmt.Errorf("LimitRange.Namespace does not match the provided context"))
	}

	oldObj, err := rs.registry.Get(ctx, limitRange.Name)
	if err != nil {
		return nil, false, err
	}

	editLimitRange := oldObj.(*api.LimitRange)

	// set the editable fields on the existing object
	editLimitRange.Labels = limitRange.Labels
	editLimitRange.ResourceVersion = limitRange.ResourceVersion
	editLimitRange.Annotations = limitRange.Annotations
	editLimitRange.Spec = limitRange.Spec

	if errs := validation.ValidateLimitRange(editLimitRange); len(errs) > 0 {
		return nil, false, errors.NewInvalid("limitRange", editLimitRange.Name, errs)
	}

	if err := rs.registry.UpdateWithName(ctx, editLimitRange.Name, editLimitRange); err != nil {
		return nil, false, err
	}
	out, err := rs.registry.Get(ctx, editLimitRange.Name)
	return out, false, err
}