Add test for atomic upgrade and install when it is interrupted

Add tests when the install release is Interrupted and the flag Wait or Atomic is set

Signed-off-by: Stephane Moser <moser.sts@gmail.com>

pkg/action/install_test.go

@@ -21,10 +21,12 @@ import (
 	"io/ioutil"
 	"log"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"regexp"
 	"strings"
 	"testing"
+	gotime "time"
 
 	"github.com/stretchr/testify/assert"
 
@@ -361,7 +363,62 @@ func TestInstallRelease_Wait(t *testing.T) {
 	is.Contains(res.Info.Description, "I timed out")
 	is.Equal(res.Info.Status, release.StatusFailed)
 }
+func TestInstallRelease_Wait_Interrupted(t *testing.T) {
+	if os.Getenv("HANDLE_SIGINT") == "1" {
+		t.Run("Execute TestInstallRelease_Wait_Interrupted", func(t *testing.T) {
+			is := assert.New(t)
+			instAction := installAction(t)
+			instAction.ReleaseName = "interrupted-release"
+			failer := instAction.cfg.KubeClient.(*kubefake.FailingKubeClient)
+			failer.PrintingKubeClient.WaitDuration = 10 * gotime.Second
+			instAction.cfg.KubeClient = failer
+			instAction.Wait = true
+			vals := map[string]interface{}{}
+
+			res, err := instAction.Run(buildChart(), vals)
+			is.Error(err)
+			is.Contains(res.Info.Description, "SIGTERM or SIGINT received, release failed")
+			is.Equal(res.Info.Status, release.StatusFailed)
+		})
+		return
+
+	}
+	t.Run("Setup TestInstallRelease_Wait_Interrupted", func(t *testing.T) {
+		cmd := exec.Command(os.Args[0], "-test.run=TestInstallRelease_Wait_Interrupted")
+		cmd.Env = append(os.Environ(), "HANDLE_SIGINT=1")
+		stdout, err := cmd.StdoutPipe()
+		if err != nil {
+			t.Fatal(err)
+		}
+		stderr, err := cmd.StderrPipe()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := cmd.Start(); err != nil {
+			t.Fatal(err)
+		}
+		go func() {
+			slurp, _ := ioutil.ReadAll(stdout)
+			fmt.Printf("%s\n", slurp)
+		}()
 
+		go func() {
+			slurp, _ := ioutil.ReadAll(stderr)
+			fmt.Printf("%s\n", slurp)
+		}()
+
+		gotime.Sleep(2 * gotime.Second)
+		p, _ := os.FindProcess(cmd.Process.Pid)
+
+		if err := p.Signal(os.Interrupt); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := cmd.Wait(); err != nil {
+			t.FailNow()
+		}
+	})
+}
 func TestInstallRelease_WaitForJobs(t *testing.T) {
 	is := assert.New(t)
 	instAction := installAction(t)
@@ -419,7 +476,69 @@ func TestInstallRelease_Atomic(t *testing.T) {
 		is.Contains(err.Error(), "an error occurred while uninstalling the release")
 	})
 }
+func TestInstallRelease_Atomic_Interrupted(t *testing.T) {
+	if os.Getenv("HANDLE_SIGINT") == "1" {
+		t.Run("Execute TestInstallRelease_Atomic_Interrupted", func(t *testing.T) {
+			is := assert.New(t)
+			instAction := installAction(t)
+			instAction.ReleaseName = "interrupted-release"
+			failer := instAction.cfg.KubeClient.(*kubefake.FailingKubeClient)
+			failer.PrintingKubeClient.WaitDuration = 10 * gotime.Second
+			instAction.cfg.KubeClient = failer
+			instAction.Atomic = true
+			vals := map[string]interface{}{}
+
+			res, err := instAction.Run(buildChart(), vals)
+			is.Error(err)
+			is.Contains(err.Error(), "SIGTERM or SIGINT received, release failed")
+			is.Contains(err.Error(), "atomic")
+			is.Contains(err.Error(), "uninstalled")
+
+			// Now make sure it isn't in storage any more
+			_, err = instAction.cfg.Releases.Get(res.Name, res.Version)
+			is.Error(err)
+			is.Equal(err, driver.ErrReleaseNotFound)
 
+		})
+		return
+
+	}
+	t.Run("Setup TestInstallRelease_Atomic_Interrupted", func(t *testing.T) {
+		cmd := exec.Command(os.Args[0], "-test.run=TestInstallRelease_Atomic_Interrupted")
+		cmd.Env = append(os.Environ(), "HANDLE_SIGINT=1")
+		stdout, err := cmd.StdoutPipe()
+		if err != nil {
+			t.Fatal(err)
+		}
+		stderr, err := cmd.StderrPipe()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := cmd.Start(); err != nil {
+			t.Fatal(err)
+		}
+		go func() {
+			slurp, _ := ioutil.ReadAll(stdout)
+			fmt.Printf("%s\n", slurp)
+		}()
+
+		go func() {
+			slurp, _ := ioutil.ReadAll(stderr)
+			fmt.Printf("%s\n", slurp)
+		}()
+
+		gotime.Sleep(2 * gotime.Second)
+		p, _ := os.FindProcess(cmd.Process.Pid)
+
+		if err := p.Signal(os.Interrupt); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := cmd.Wait(); err != nil {
+			t.FailNow()
+		}
+	})
+}
 func TestNameTemplate(t *testing.T) {
 	testCases := []nameTemplateTestCase{
 		// Just a straight up nop please

pkg/action/upgrade.go

@@ -325,6 +325,9 @@ func (u *Upgrade) performUpgrade(originalRelease, upgradedRelease *release.Relea
 	return result.r, result.e
 }
 
+// Function used to lock the Mutex, this is important for the case when the atomic flag is set.
+// In that case the upgrade will finish before the rollback is finished so it is necessary to wait for the rollback to finish.
+// The rollback will be trigger by the function failRelease
 func (u *Upgrade) reportToPerformUpgrade(c chan<- resultMessage, rel *release.Release, created kube.ResourceList, err error) {
 	upgradeLock.Lock()
 	if err != nil {
@@ -333,15 +336,15 @@ func (u *Upgrade) reportToPerformUpgrade(c chan<- resultMessage, rel *release.Re
 	c <- resultMessage{r: rel, e: err}
 	upgradeLock.Unlock()
 }
+
+// Setup listener for SIGINT and SIGTERM
 func (u *Upgrade) handleSignals(c chan<- resultMessage, upgradedRelease *release.Release) {
-	// Handle SIGINT
 	cSignal := make(chan os.Signal)
 	signal.Notify(cSignal, os.Interrupt, syscall.SIGTERM)
 	go func() {
 		<-cSignal
 		u.cfg.Log("SIGTERM or SIGINT received")
-		// when the atomic flag is set the ongoing release finish first and doesn't give time for the rollback happens . I need to think in a way to lock the chanel
-		// Implement function reportToPerformUpgrade(channel, Release, ResourceList, error) if error != nill call u.failRelease
+		// when the atomic flag is set the ongoing release finish first and doesn't give time for the rollback happens.
 		u.reportToPerformUpgrade(c, upgradedRelease, kube.ResourceList{}, fmt.Errorf("SIGTERM or SIGINT received, release failed"))
 	}()
 }

pkg/action/upgrade_test.go

@@ -363,3 +363,71 @@ func TestUpgradeRelease_Interrupted_Wait(t *testing.T) {
 		}
 	})
 }
+
+func TestUpgradeRelease_Interrupted_Atomic(t *testing.T) {
+	if os.Getenv("HANDLE_SIGINT") == "1" {
+		t.Run("Execute TestUpgradeRelease_Interrupted_Atomic", func(t *testing.T) {
+			is := assert.New(t)
+			req := require.New(t)
+
+			upAction := upgradeAction(t)
+			rel := releaseStub()
+			rel.Name = "interrupted-release"
+			rel.Info.Status = release.StatusDeployed
+			upAction.cfg.Releases.Create(rel)
+
+			failer := upAction.cfg.KubeClient.(*kubefake.FailingKubeClient)
+			failer.PrintingKubeClient.WaitDuration = 5 * gotime.Second
+			upAction.cfg.KubeClient = failer
+			upAction.Atomic = true
+			vals := map[string]interface{}{}
+			res, err := upAction.Run(rel.Name, buildChart(), vals)
+
+			req.Error(err)
+			is.Contains(err.Error(), "release interrupted-release failed, and has been rolled back due to atomic being set: SIGTERM or SIGINT received, release failed")
+
+			// Now make sure it is actually upgraded
+			updatedRes, err := upAction.cfg.Releases.Get(res.Name, 3)
+			is.NoError(err)
+			// Should have rolled back to the previous
+			is.Equal(updatedRes.Info.Status, release.StatusDeployed)
+		})
+		return
+
+	}
+	t.Run("Setup TestUpgradeRelease_Interrupted_Atomic", func(t *testing.T) {
+		cmd := exec.Command(os.Args[0], "-test.run=TestUpgradeRelease_Interrupted_Atomic")
+		cmd.Env = append(os.Environ(), "HANDLE_SIGINT=1")
+		stdout, err := cmd.StdoutPipe()
+		if err != nil {
+			t.Fatal(err)
+		}
+		stderr, err := cmd.StderrPipe()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := cmd.Start(); err != nil {
+			t.Fatal(err)
+		}
+		go func() {
+			slurp, _ := ioutil.ReadAll(stdout)
+			fmt.Printf("%s\n", slurp)
+		}()
+
+		go func() {
+			slurp, _ := ioutil.ReadAll(stderr)
+			fmt.Printf("%s\n", slurp)
+		}()
+
+		gotime.Sleep(2 * gotime.Second)
+		p, _ := os.FindProcess(cmd.Process.Pid)
+
+		if err := p.Signal(os.Interrupt); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := cmd.Wait(); err != nil {
+			t.FailNow()
+		}
+	})
+}