fix(tiller): improve handling of corrupted storage

For some reason, many users experince corrupted storage with the
ConfigMaps storage backend. Specifically, several Releases are marked as
DEPLOYED. This patch improved handling of such situations, by taking the latest
DEPLOYED Release. Eventually, the storage will clean itself out, after
the corrupted Releases are deleted due to --history-max.

Closes #6031

Signed-off-by: Cristian Klein <cristian.klein@elastisys.com>

pkg/storage/storage.go

@@ -136,6 +136,9 @@ func (s *Storage) Deployed(name string) (*rspb.Release, error) {
 		return nil, fmt.Errorf("%q %s", name, NoReleasesErr)
 	}
 
+	// Sometimes Tiller's database gets corrupted and multiple releases are DEPLOYED. Take the latest.
+	relutil.Reverse(ls, relutil.SortByRevision)
+
 	return ls[0], err
 }
 

pkg/storage/storage_test.go

@@ -205,6 +205,46 @@ func TestStorageDeployed(t *testing.T) {
 	}
 }
 
+func TestStorageDeployedWithCorruption(t *testing.T) {
+	storage := Init(driver.NewMemory())
+
+	const name = "angry-bird"
+	const vers = int32(4)
+
+	// setup storage with test releases
+	setup := func() {
+		// release records (notice odd order and corruption)
+		rls0 := ReleaseTestData{Name: name, Version: 1, Status: rspb.Status_SUPERSEDED}.ToRelease()
+		rls1 := ReleaseTestData{Name: name, Version: 4, Status: rspb.Status_DEPLOYED}.ToRelease()
+		rls2 := ReleaseTestData{Name: name, Version: 3, Status: rspb.Status_SUPERSEDED}.ToRelease()
+		rls3 := ReleaseTestData{Name: name, Version: 2, Status: rspb.Status_DEPLOYED}.ToRelease()
+
+		// create the release records in the storage
+		assertErrNil(t.Fatal, storage.Create(rls0), "Storing release 'angry-bird' (v1)")
+		assertErrNil(t.Fatal, storage.Create(rls1), "Storing release 'angry-bird' (v2)")
+		assertErrNil(t.Fatal, storage.Create(rls2), "Storing release 'angry-bird' (v3)")
+		assertErrNil(t.Fatal, storage.Create(rls3), "Storing release 'angry-bird' (v4)")
+	}
+
+	setup()
+
+	rls, err := storage.Deployed(name)
+	if err != nil {
+		t.Fatalf("Failed to query for deployed release: %s\n", err)
+	}
+
+	switch {
+	case rls == nil:
+		t.Fatalf("Release is nil")
+	case rls.Name != name:
+		t.Fatalf("Expected release name %q, actual %q\n", name, rls.Name)
+	case rls.Version != vers:
+		t.Fatalf("Expected release version %d, actual %d\n", vers, rls.Version)
+	case rls.Info.Status.Code != rspb.Status_DEPLOYED:
+		t.Fatalf("Expected release status 'DEPLOYED', actual %s\n", rls.Info.Status.Code)
+	}
+}
+
 func TestStorageHistory(t *testing.T) {
 	storage := Init(driver.NewMemory())