metadata: ensure correct updates on Container

This fixes a few bugs in the container store related to reading and
writing fields. Specifically, on update, the full field set wasn't being
returned to the caller, making it appear that the store was corrupted.
We now return the correctly updated field and store the missing field
that was omitted in the original implementation. In course, we also have
defined the update semantics of each field, as well as whether or not
they are required.

The big addition here is really the container metadata testsuite. It
covers listing, filtering, creates, updates and deletes in a vareity of
scenarios.

Signed-off-by: Stephen J Day <stephen.day@docker.com>

metadata/containers.go

@@ -91,8 +91,8 @@ func (s *containerStore) Create(ctx context.Context, container containers.Contai
 		return containers.Container{}, err
 	}
 
-	if err := identifiers.Validate(container.ID); err != nil {
-		return containers.Container{}, err
+	if err := validateContainer(&container); err != nil {
+		return containers.Container{}, errors.Wrap(err, "create container failed validation")
 	}
 
 	bkt, err := createContainersBucket(s.tx, namespace)
@@ -144,37 +144,55 @@ func (s *containerStore) Update(ctx context.Context, container containers.Contai
 	createdat := updated.CreatedAt
 	updated.ID = container.ID
 
+	if len(fieldpaths) == 0 {
+		// only allow updates to these field on full replace.
+		fieldpaths = []string{"labels", "spec"}
+
+		// Fields that are immutable must cause an error when no field paths
+		// are provided. This allows these fields to become mutable in the
+		// future.
+		if updated.Image != container.Image {
+			return containers.Container{}, errors.Wrapf(errdefs.ErrInvalidArgument, "container.Image field is immutable")
+		}
+
+		if updated.RootFS != container.RootFS {
+			return containers.Container{}, errors.Wrapf(errdefs.ErrInvalidArgument, "container.RootFS field is immutable")
+		}
+
+		if updated.Snapshotter != container.Snapshotter {
+			return containers.Container{}, errors.Wrapf(errdefs.ErrInvalidArgument, "container.Snapshotter field is immutable")
+		}
+
+		if updated.Runtime.Name != container.Runtime.Name {
+			return containers.Container{}, errors.Wrapf(errdefs.ErrInvalidArgument, "container.Runtime.Name field is immutable")
+		}
+	}
+
 	// apply the field mask. If you update this code, you better follow the
 	// field mask rules in field_mask.proto. If you don't know what this
 	// is, do not update this code.
-	if len(fieldpaths) > 0 {
-		// TODO(stevvooe): Move this logic into the store itself.
-		for _, path := range fieldpaths {
-			if strings.HasPrefix(path, "labels.") {
-				if updated.Labels == nil {
-					updated.Labels = map[string]string{}
-				}
-				key := strings.TrimPrefix(path, "labels.")
-				updated.Labels[key] = container.Labels[key]
-				continue
-			}
-
-			switch path {
-			case "labels":
-				updated.Labels = container.Labels
-			case "image":
-				updated.Image = container.Image
-			case "spec":
-				updated.Spec = container.Spec
-			case "rootfs":
-				updated.RootFS = container.RootFS
-			default:
-				return containers.Container{}, errors.Wrapf(errdefs.ErrInvalidArgument, "cannot update %q field on %q", path, container.ID)
+	for _, path := range fieldpaths {
+		if strings.HasPrefix(path, "labels.") {
+			if updated.Labels == nil {
+				updated.Labels = map[string]string{}
 			}
+			key := strings.TrimPrefix(path, "labels.")
+			updated.Labels[key] = container.Labels[key]
+			continue
 		}
-	} else {
-		// no field mask present, just replace everything
-		updated = container
+
+		switch path {
+		case "labels":
+			updated.Labels = container.Labels
+		case "spec":
+			updated.Spec = container.Spec
+		default:
+			return containers.Container{}, errors.Wrapf(errdefs.ErrInvalidArgument, "cannot update %q field on %q", path, container.ID)
+		}
+	}
+
+	if err := validateContainer(&updated); err != nil {
+		return containers.Container{}, errors.Wrap(err, "update failed validation")
 	}
 
 	updated.CreatedAt = createdat
@@ -183,7 +201,7 @@ func (s *containerStore) Update(ctx context.Context, container containers.Contai
 		return containers.Container{}, errors.Wrap(err, "failed to write container")
 	}
 
-	return container, nil
+	return updated, nil
 }
 
 func (s *containerStore) Delete(ctx context.Context, id string) error {
@@ -203,6 +221,27 @@ func (s *containerStore) Delete(ctx context.Context, id string) error {
 	return err
 }
 
+func validateContainer(container *containers.Container) error {
+	if err := identifiers.Validate(container.ID); err != nil {
+		return errors.Wrapf(err, "container.ID validation error")
+	}
+
+	// labels and image have no validation
+	if container.Runtime.Name == "" {
+		return errors.Wrapf(errdefs.ErrInvalidArgument, "container.Runtime.Name must be set")
+	}
+
+	if container.Spec == nil {
+		return errors.Wrapf(errdefs.ErrInvalidArgument, "container.Spec must be set")
+	}
+
+	if container.RootFS != "" && container.Snapshotter == "" {
+		return errors.Wrapf(errdefs.ErrInvalidArgument, "container.Snapshotter must be set if container.RootFS is set")
+	}
+
+	return nil
+}
+
 func readContainer(container *containers.Container, bkt *bolt.Bucket) error {
 	labels, err := boltutil.ReadLabels(bkt)
 	if err != nil {
@@ -247,7 +286,8 @@ func readContainer(container *containers.Container, bkt *bolt.Bucket) error {
 			container.Spec = &any
 		case string(bucketKeyRootFS):
 			container.RootFS = string(v)
-
+		case string(bucketKeySnapshotter):
+			container.Snapshotter = string(v)
 		}
 
 		return nil
@@ -272,6 +312,7 @@ func writeContainer(bkt *bolt.Bucket, container *containers.Container) error {
 
 	for _, v := range [][2][]byte{
 		{bucketKeyImage, []byte(container.Image)},
+		{bucketKeySnapshotter, []byte(container.Snapshotter)},
 		{bucketKeyRootFS, []byte(container.RootFS)},
 	} {
 		if err := bkt.Put(v[0], v[1]); err != nil {