Verify container changes

cluster/kubernetes/resource/cronjob.go

@@ -1,6 +1,7 @@
 package resource
 
 import (
+	"github.com/weaveworks/flux/image"
 	"github.com/weaveworks/flux/resource"
 )
 
@@ -20,3 +21,9 @@ type CronJobSpec struct {
 func (c CronJob) Containers() []resource.Container {
 	return c.Spec.JobTemplate.Spec.Template.Containers()
 }
+
+func (c CronJob) SetContainerImage(container string, ref image.Ref) error {
+	return c.Spec.JobTemplate.Spec.Template.SetContainerImage(container, ref)
+}
+
+var _ resource.Workload = CronJob{}

cluster/kubernetes/resource/daemonset.go

@@ -1,18 +1,23 @@
 package resource
 
 import (
+	"github.com/weaveworks/flux/image"
 	"github.com/weaveworks/flux/resource"
 )
 
 type DaemonSet struct {
 	baseObject
-	Spec DaemonSetSpec
-}
-
-type DaemonSetSpec struct {
-	Template PodTemplate
+	Spec struct {
+		Template PodTemplate
+	}
 }
 
 func (ds DaemonSet) Containers() []resource.Container {
 	return ds.Spec.Template.Containers()
 }
+
+func (ds DaemonSet) SetContainerImage(container string, ref image.Ref) error {
+	return ds.Spec.Template.SetContainerImage(container, ref)
+}
+
+var _ resource.Workload = DaemonSet{}

cluster/kubernetes/resource/deployment.go

@@ -1,6 +1,7 @@
 package resource
 
 import (
+	"github.com/weaveworks/flux/image"
 	"github.com/weaveworks/flux/resource"
 )
 
@@ -17,3 +18,9 @@ type DeploymentSpec struct {
 func (d Deployment) Containers() []resource.Container {
 	return d.Spec.Template.Containers()
 }
+
+func (d Deployment) SetContainerImage(container string, ref image.Ref) error {
+	return d.Spec.Template.SetContainerImage(container, ref)
+}
+
+var _ resource.Workload = Deployment{}

cluster/kubernetes/resource/spec.go

@@ -1,6 +1,8 @@
 package resource
 
 import (
+	"fmt"
+
 	"github.com/weaveworks/flux/image"
 	"github.com/weaveworks/flux/resource"
 )
@@ -28,6 +30,16 @@ func (t PodTemplate) Containers() []resource.Container {
 	return result
 }
 
+func (t PodTemplate) SetContainerImage(container string, ref image.Ref) error {
+	for i, c := range t.Spec.Containers {
+		if c.Name == container {
+			t.Spec.Containers[i].Image = ref.String()
+			return nil
+		}
+	}
+	return fmt.Errorf("container %q not found in workload", container)
+}
+
 type PodSpec struct {
 	ImagePullSecrets []struct{ Name string }
 	Volumes          []Volume

cluster/kubernetes/resource/statefulset.go

@@ -1,6 +1,7 @@
 package resource
 
 import (
+	"github.com/weaveworks/flux/image"
 	"github.com/weaveworks/flux/resource"
 )
 
@@ -17,3 +18,9 @@ type StatefulSetSpec struct {
 func (ss StatefulSet) Containers() []resource.Container {
 	return ss.Spec.Template.Containers()
 }
+
+func (ss StatefulSet) SetContainerImage(container string, ref image.Ref) error {
+	return ss.Spec.Template.SetContainerImage(container, ref)
+}
+
+var _ resource.Workload = StatefulSet{}

release/errors.go

@@ -0,0 +1,30 @@
+package release
+
+import (
+	fluxerr "github.com/weaveworks/flux/errors"
+)
+
+func MakeReleaseError(err error) *fluxerr.Error {
+	return &fluxerr.Error{
+		Type: fluxerr.User,
+		Help: `The release process failed, with this message:
+
+    ` + err.Error() + `
+
+This may be because of a limitation in the formats of file Flux can
+deal with. See
+
+    https://github.com/weaveworks/flux/blob/master/site/requirements.md
+
+for those limitations.
+
+If your files appear to meet the requirements, it may simply be a bug
+in Flux. Please report it at
+
+    https://github.com/weaveworks/flux/issues
+
+and try to include the problematic manifest, if it can be identified.
+`,
+		Err: err,
+	}
+}

release/releaser.go

@@ -1,10 +1,14 @@
 package release
 
 import (
+	"fmt"
+	"strings"
 	"time"
 
 	"github.com/go-kit/kit/log"
+	"github.com/pkg/errors"
 
+	"github.com/weaveworks/flux/resource"
 	"github.com/weaveworks/flux/update"
 )
 
@@ -27,12 +31,26 @@ func Release(rc *ReleaseContext, changes Changes, logger log.Logger) (results up
 
 	logger = log.With(logger, "type", "release")
 
+	before, err := rc.manifests.LoadManifests(rc.repo.Dir(), rc.repo.ManifestDir())
 	updates, results, err := changes.CalculateRelease(rc, logger)
 	if err != nil {
 		return nil, err
 	}
 
 	err = ApplyChanges(rc, updates, logger)
+	if err != nil {
+		return results, MakeReleaseError(errors.Wrap(err, "applying changes"))
+	}
+
+	after, err := rc.manifests.LoadManifests(rc.repo.Dir(), rc.repo.ManifestDir())
+	if err != nil {
+		return results, MakeReleaseError(errors.Wrap(err, "loading resources after updates"))
+	}
+
+	err = VerifyChanges(before, updates, after)
+	if err != nil {
+		return results, MakeReleaseError(errors.Wrap(err, "verifying changes"))
+	}
 	return results, err
 }
 
@@ -48,3 +66,71 @@ func ApplyChanges(rc *ReleaseContext, updates []*update.ControllerUpdate, logger
 	timer.ObserveDuration()
 	return err
 }
+
+// VerifyChanges checks that the `after` resources are exactly the
+// `before` resources with the updates applied. It destructively
+// updates `before`.
+func VerifyChanges(before map[string]resource.Resource, updates []*update.ControllerUpdate, after map[string]resource.Resource) error {
+	timer := update.NewStageTimer("verify_changes")
+	defer timer.ObserveDuration()
+
+	verificationError := func(msg string, args ...interface{}) error {
+		return errors.Wrap(fmt.Errorf(msg, args...), "failed to verify changes")
+	}
+
+	for _, update := range updates {
+		res, ok := before[update.ResourceID.String()]
+		if !ok {
+			return verificationError("resource %q mentioned in update not found in resources", update.ResourceID.String())
+		}
+		wl, ok := res.(resource.Workload)
+		if !ok {
+			return verificationError("resource %q mentioned in update is not a workload", update.ResourceID.String())
+		}
+		for _, containerUpdate := range update.Updates {
+			wl.SetContainerImage(containerUpdate.Container, containerUpdate.Target)
+		}
+	}
+
+	for id, afterRes := range after {
+		beforeRes, ok := before[id]
+		if !ok {
+			return verificationError("resource %q is new after update")
+		}
+		delete(before, id)
+
+		beforeWorkload, ok := beforeRes.(resource.Workload)
+		if !ok {
+			// the resource in question isn't a workload, so ignore it.
+			continue
+		}
+		afterWorkload, ok := afterRes.(resource.Workload)
+		if !ok {
+			return verificationError("resource %q is no longer a workload (Deployment or DaemonSet, or similar) after update", id)
+		}
+
+		beforeContainers := beforeWorkload.Containers()
+		afterContainers := afterWorkload.Containers()
+		if len(beforeContainers) != len(afterContainers) {
+			return verificationError("resource %q has different set of containers after update", id)
+		}
+		for i := range afterContainers {
+			if beforeContainers[i].Name != afterContainers[i].Name {
+				return verificationError("container in position %d of resource %q has a different name after update: was %q, now %q", i, id, beforeContainers[i].Name, afterContainers[i].Name)
+			}
+			if beforeContainers[i].Image != afterContainers[i].Image {
+				return verificationError("the image for container %q in resource %q should be %q, but is %q", beforeContainers[i].Name, id, beforeContainers[i].Image.String(), afterContainers[i].Image.String())
+			}
+		}
+	}
+
+	var disappeared []string
+	for id := range before {
+		disappeared = append(disappeared, fmt.Sprintf("%q", id))
+	}
+	if len(disappeared) > 0 {
+		return verificationError("resources {%s} present before update but not after", strings.Join(disappeared, ", "))
+	}
+
+	return nil
+}

release/releaser_test.go

@@ -439,3 +439,51 @@ func testRelease(t *testing.T, name string, ctx *ReleaseContext, spec update.Rel
 		t.Errorf("%s\n--- expected ---\n%s\n--- got ---\n%s\n", name, string(exp), string(got))
 	}
 }
+
+// --- test verification
+
+// A Manifests implementation that does updates incorrectly, so they should fail verification.
+type badManifests struct {
+	kubernetes.Manifests
+}
+
+func (m *badManifests) UpdateImage(def []byte, resourceID flux.ResourceID, container string, newImageID image.Ref) ([]byte, error) {
+	return def, nil
+}
+
+func TestBadRelease(t *testing.T) {
+	cluster := mockCluster(hwSvc)
+	spec := update.ReleaseSpec{
+		ServiceSpecs: []update.ResourceSpec{update.ResourceSpecAll},
+		ImageSpec:    update.ImageSpecFromRef(newHwRef),
+		Kind:         update.ReleaseKindExecute,
+		Excludes:     []flux.ResourceID{},
+	}
+	checkout1, cleanup1 := setup(t)
+	defer cleanup1()
+
+	ctx := &ReleaseContext{
+		cluster:   cluster,
+		manifests: &kubernetes.Manifests{},
+		repo:      checkout1,
+		registry:  mockRegistry,
+	}
+	_, err := Release(ctx, spec, log.NewNopLogger())
+	if err != nil {
+		t.Fatal("release with 'good' Manifests should succeed, but errored:", err)
+	}
+
+	checkout2, cleanup2 := setup(t)
+	defer cleanup2()
+
+	ctx = &ReleaseContext{
+		cluster:   cluster,
+		manifests: &badManifests{Manifests: kubernetes.Manifests{}},
+		repo:      checkout2,
+		registry:  mockRegistry,
+	}
+	_, err = Release(ctx, spec, log.NewNopLogger())
+	if err == nil {
+		t.Fatal("did not return an error, but was expected to fail verification")
+	}
+}

resource/resource.go

@@ -22,4 +22,8 @@ type Container struct {
 type Workload interface {
 	Resource
 	Containers() []Container
+	// SetContainerImage mutates this workload so that the container
+	// named has the image given. This is not expected to have an
+	// effect on any underlying file or cluster resource.
+	SetContainerImage(container string, ref image.Ref) error
 }