fix(tiller): Supersede multiple deployments

There are cases when multiple revisions of a release has been
marked with DEPLOYED status. This makes sure any previous deployment
will be set to SUPERSEDED when doing rollbacks.

Closes #2941 #3513 #3275

pkg/storage/storage.go

@@ -98,7 +98,7 @@ func (s *Storage) ListDeployed() ([]*rspb.Release, error) {
 	})
 }
 
-// ListFilterAll returns the set of releases satisfying satisfying the predicate
+// ListFilterAll returns the set of releases satisfying the predicate
 // (filter0 && filter1 && ... && filterN), i.e. a Release is included in the results
 // if and only if all filters return true.
 func (s *Storage) ListFilterAll(fns ...relutil.FilterFunc) ([]*rspb.Release, error) {
@@ -108,7 +108,7 @@ func (s *Storage) ListFilterAll(fns ...relutil.FilterFunc) ([]*rspb.Release, err
 	})
 }
 
-// ListFilterAny returns the set of releases satisfying satisfying the predicate
+// ListFilterAny returns the set of releases satisfying the predicate
 // (filter0 || filter1 || ... || filterN), i.e. a Release is included in the results
 // if at least one of the filters returns true.
 func (s *Storage) ListFilterAny(fns ...relutil.FilterFunc) ([]*rspb.Release, error) {
@@ -118,10 +118,24 @@ func (s *Storage) ListFilterAny(fns ...relutil.FilterFunc) ([]*rspb.Release, err
 	})
 }
 
-// Deployed returns the deployed release with the provided release name, or
+// Deployed returns the last deployed release with the provided release name, or
 // returns ErrReleaseNotFound if not found.
 func (s *Storage) Deployed(name string) (*rspb.Release, error) {
-	s.Log("getting deployed release from %q history", name)
+	ls, err := s.DeployedAll(name)
+	if err != nil {
+		if strings.Contains(err.Error(), "not found") {
+			return nil, fmt.Errorf("%q has no deployed releases", name)
+		}
+		return nil, err
+	}
+
+	return ls[0], err
+}
+
+// DeployedAll returns all deployed releases with the provided name, or
+// returns ErrReleaseNotFound if not found.
+func (s *Storage) DeployedAll(name string) ([]*rspb.Release, error) {
+	s.Log("getting deployed releases from %q history", name)
 
 	ls, err := s.Driver.Query(map[string]string{
 		"NAME":   name,
@@ -129,7 +143,7 @@ func (s *Storage) Deployed(name string) (*rspb.Release, error) {
 		"STATUS": "DEPLOYED",
 	})
 	if err == nil {
-		return ls[0], nil
+		return ls, nil
 	}
 	if strings.Contains(err.Error(), "not found") {
 		return nil, fmt.Errorf("%q has no deployed releases", name)

pkg/tiller/release_rollback.go

@@ -146,8 +146,16 @@ func (s *ReleaseServer) performRollback(currentRelease, targetRelease *release.R
 		}
 	}
 
-	currentRelease.Info.Status.Code = release.Status_SUPERSEDED
+	deployed, err := s.env.Releases.DeployedAll(currentRelease.Name)
-	s.recordRelease(currentRelease, true)
+	if err != nil {
+		return nil, err
+	}
+	// Supersede all previous deployments, see issue #2941.
+	for _, r := range deployed {
+		s.Log("superseding previous deployment %d", r.Version)
+		r.Info.Status.Code = release.Status_SUPERSEDED
+		s.recordRelease(r, true)
+	}
 
 	targetRelease.Info.Status.Code = release.Status_DEPLOYED
 

pkg/tiller/release_server.go

@@ -316,6 +316,7 @@ func (s *ReleaseServer) renderResources(ch *chart.Chart, values chartutil.Values
 	return hooks, b, notes, nil
 }
 
+// recordRelease with an update operation in case reuse has been set.
 func (s *ReleaseServer) recordRelease(r *release.Release, reuse bool) {
 	if reuse {
 		if err := s.env.Releases.Update(r); err != nil {