Upgrade EC managers

pkg/handlers/websocket.go

@@ -23,7 +23,15 @@ func (h *Handler) ConnectToECWebsocket(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if err := websocket.Connect(w, r, nodeName); err != nil {
+	version := r.URL.Query().Get("version")
+	if version == "" {
+		response.Error = "missing version"
+		logger.Error(errors.New(response.Error))
+		JSON(w, http.StatusBadRequest, response)
+		return
+	}
+
+	if err := websocket.Connect(w, r, nodeName, version); err != nil {
 		response.Error = "failed to establish websocket connection"
 		logger.Error(errors.Wrap(err, response.Error))
 		JSON(w, http.StatusInternalServerError, response)

pkg/plan/app.go

@@ -18,9 +18,47 @@ import (
 	"github.com/replicatedhq/kots/pkg/upgradeservice"
 	upgradeservicetypes "github.com/replicatedhq/kots/pkg/upgradeservice/types"
 	"github.com/replicatedhq/kots/pkg/util"
+	"github.com/segmentio/ksuid"
 )
 
+func planAppUpgradeService(s store.Store, p *types.Plan) ([]*types.PlanStep, error) {
+	ausInput, err := getAppUpgradeServiceInput(s, p, ksuid.New().String())
+	if err != nil {
+		return nil, errors.Wrap(err, "get app upgrade service input")
+	}
+	steps := []*types.PlanStep{
+		{
+			ID:                ausInput.Params.PlanStepID,
+			Name:              "App Upgrade Service",
+			Type:              types.StepTypeAppUpgradeService,
+			Status:            types.StepStatusPending,
+			StatusDescription: "Pending",
+			Owner:             types.StepOwnerKOTS,
+			Input:             *ausInput,
+		},
+	}
+	return steps, nil
+}
+
+func planAppUpgrade() ([]*types.PlanStep, error) {
+	return []*types.PlanStep{
+		{
+			ID:                ksuid.New().String(),
+			Name:              "Application Upgrade",
+			Type:              types.StepTypeAppUpgrade,
+			Status:            types.StepStatusPending,
+			StatusDescription: "Pending application upgrade",
+			Owner:             types.StepOwnerKOTS,
+			// the input here is the app upgrade service output
+		},
+	}, nil
+}
+
 func executeAppUpgradeService(s store.Store, p *types.Plan, step *types.PlanStep) (finalError error) {
+	if step.Status != types.StepStatusPending {
+		return errors.Errorf("step %q cannot be resumed", step.Name)
+	}
+
 	in, ok := step.Input.(types.PlanStepInputAppUpgradeService)
 	if !ok {
 		return errors.New("invalid input for app upgrade service step")
@@ -50,6 +88,10 @@ func executeAppUpgradeService(s store.Store, p *types.Plan, step *types.PlanStep
 		return errors.Wrap(err, "update step status")
 	}
 
+	if err := waitForStep(p, step.ID); err != nil {
+		return errors.Wrap(err, "wait for upgrade service")
+	}
+
 	return nil
 }
 

pkg/plan/cluster.go

@@ -13,42 +13,75 @@ import (
 	"github.com/replicatedhq/kots/pkg/plan/types"
 	"github.com/replicatedhq/kots/pkg/store"
 	"github.com/replicatedhq/kots/pkg/websocket"
+	"github.com/segmentio/ksuid"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
 	k8syaml "sigs.k8s.io/yaml"
 )
 
-func executeECUpgrade(s store.Store, p *types.Plan, step *types.PlanStep) error {
-	in, ok := step.Input.(types.PlanStepInputECUpgrade)
+func planK0sUpgrade(s store.Store, kcli kbclient.Client, a *apptypes.App, versionLabel string, newSpec *ecv1beta1.ConfigSpec) ([]*types.PlanStep, error) {
+	steps := []*types.PlanStep{}
+
+	requiresUpgrade, err := requiresK0sUpgrade(kcli, newSpec)
+	if err != nil {
+		return nil, errors.Wrap(err, "check if requires k0s upgrade")
+	}
+	if requiresUpgrade {
+		in, err := getK0sUpgradeInput(s, kcli, a, versionLabel, newSpec)
+		if err != nil {
+			return nil, errors.Wrap(err, "get k0s upgrade input")
+		}
+		steps = append(steps, &types.PlanStep{
+			ID:                ksuid.New().String(),
+			Name:              "K0s Upgrade",
+			Type:              types.StepTypeK0sUpgrade,
+			Status:            types.StepStatusPending,
+			StatusDescription: "Pending K0s upgrade",
+			Input:             *in,
+			Owner:             types.StepOwnerECManager,
+		})
+	}
+
+	return steps, nil
+}
+
+func executeK0sUpgrade(s store.Store, p *types.Plan, step *types.PlanStep) error {
+	in, ok := step.Input.(types.PlanStepInputK0sUpgrade)
 	if !ok {
-		return errors.New("invalid input for embedded cluster upgrade step")
+		return errors.New("invalid input for k0s upgrade step")
 	}
 
-	newInstall := &ecv1beta1.Installation{
-		TypeMeta: metav1.TypeMeta{
-			APIVersion: ecv1beta1.GroupVersion.String(),
-			Kind:       "Installation",
-		},
-		ObjectMeta: metav1.ObjectMeta{
-			Name: time.Now().Format("20060102150405"),
-			Labels: map[string]string{
-				"replicated.com/disaster-recovery": "ec-install",
+	if step.Status == types.StepStatusPending {
+		newInstall := &ecv1beta1.Installation{
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: ecv1beta1.GroupVersion.String(),
+				Kind:       "Installation",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name: time.Now().Format("20060102150405"),
+				Labels: map[string]string{
+					"replicated.com/disaster-recovery": "ec-install",
+				},
 			},
-		},
-		Spec: in.CurrentECInstallation.Spec,
+			Spec: in.CurrentECInstallation.Spec,
+		}
+		newInstall.Spec.Artifacts = embeddedcluster.GetArtifactsFromInstallation(in.CurrentKOTSInstallation)
+		newInstall.Spec.Config = &in.NewECConfigSpec
+		newInstall.Spec.LicenseInfo = &ecv1beta1.LicenseInfo{IsDisasterRecoverySupported: in.IsDisasterRecoverySupported}
+
+		if err := websocket.UpgradeCluster(newInstall, p.AppSlug, p.VersionLabel, step.ID); err != nil {
+			return errors.Wrap(err, "upgrade cluster")
+		}
 	}
-	newInstall.Spec.Artifacts = embeddedcluster.GetArtifactsFromInstallation(in.CurrentKOTSInstallation)
-	newInstall.Spec.Config = &in.NewECConfigSpec
-	newInstall.Spec.LicenseInfo = &ecv1beta1.LicenseInfo{IsDisasterRecoverySupported: in.IsDisasterRecoverySupported}
 
-	if err := websocket.UpgradeCluster(newInstall, p.AppSlug, p.VersionLabel, step.ID); err != nil {
-		return errors.Wrap(err, "upgrade cluster")
+	if err := waitForStep(p, step.ID); err != nil {
+		return errors.Wrap(err, "wait for k0s upgrade")
 	}
 
 	return nil
 }
 
-func requiresECUpgrade(kcli kbclient.Client, newSpec *ecv1beta1.ConfigSpec) (bool, error) {
+func requiresK0sUpgrade(kcli kbclient.Client, newSpec *ecv1beta1.ConfigSpec) (bool, error) {
 	currInstall, err := embeddedcluster.GetCurrentInstallation(context.Background(), kcli)
 	if err != nil {
 		return false, errors.Wrap(err, "get current embedded cluster installation")
@@ -70,7 +103,7 @@ func requiresECUpgrade(kcli kbclient.Client, newSpec *ecv1beta1.ConfigSpec) (boo
 	return false, nil
 }
 
-func getECUpgradeInput(s store.Store, kcli kbclient.Client, a *apptypes.App, versionLabel string, newSpec *ecv1beta1.ConfigSpec) (*types.PlanStepInputECUpgrade, error) {
+func getK0sUpgradeInput(s store.Store, kcli kbclient.Client, a *apptypes.App, versionLabel string, newSpec *ecv1beta1.ConfigSpec) (*types.PlanStepInputK0sUpgrade, error) {
 	license, err := kotsutil.LoadLicenseFromBytes([]byte(a.License))
 	if err != nil {
 		return nil, errors.Wrap(err, "parse app license")
@@ -91,7 +124,7 @@ func getECUpgradeInput(s store.Store, kcli kbclient.Client, a *apptypes.App, ver
 		return nil, errors.Wrap(err, "get current embedded cluster installation")
 	}
 
-	return &types.PlanStepInputECUpgrade{
+	return &types.PlanStepInputK0sUpgrade{
 		CurrentECInstallation:       *currECInstall,
 		CurrentKOTSInstallation:     *currKOTSInstall,
 		NewECConfigSpec:             *newSpec,

pkg/plan/extensions.go

@@ -9,6 +9,7 @@ import (
 	"github.com/replicatedhq/kots/pkg/embeddedcluster"
 	"github.com/replicatedhq/kots/pkg/plan/types"
 	"github.com/replicatedhq/kots/pkg/websocket"
+	"github.com/segmentio/ksuid"
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -18,14 +19,84 @@ type ExtensionsDiffResult struct {
 	Modified []ecv1beta1.Chart
 }
 
+func planECExtensions(kcli kbclient.Client, newSpec *ecv1beta1.ConfigSpec) ([]*types.PlanStep, error) {
+	steps := []*types.PlanStep{}
+
+	currECExts, newECExts, err := getECExtensions(kcli, newSpec)
+	if err != nil {
+		return nil, errors.Wrap(err, "get extensions")
+	}
+
+	ecExtsDiff := diffECExtensions(currECExts, newECExts)
+	newRepos := newECExts.Helm.Repositories
+
+	// added extensions
+	for _, chart := range ecExtsDiff.Added {
+		steps = append(steps, &types.PlanStep{
+			ID:                ksuid.New().String(),
+			Name:              "Extension Add",
+			Type:              types.StepTypeECExtensionAdd,
+			Status:            types.StepStatusPending,
+			StatusDescription: "Pending extension addition",
+			Input: types.PlanStepInputECExtension{
+				Repos: newRepos,
+				Chart: chart,
+			},
+			Owner: types.StepOwnerECManager,
+		})
+	}
+
+	// modified extensions
+	for _, chart := range ecExtsDiff.Modified {
+		steps = append(steps, &types.PlanStep{
+			ID:                ksuid.New().String(),
+			Name:              "Extension Upgrade",
+			Type:              types.StepTypeECExtensionUpgrade,
+			Status:            types.StepStatusPending,
+			StatusDescription: "Pending extension upgrade",
+			Input: types.PlanStepInputECExtension{
+				Repos: newRepos,
+				Chart: chart,
+			},
+			Owner: types.StepOwnerECManager,
+		})
+	}
+
+	// removed extensions
+	for _, chart := range ecExtsDiff.Removed {
+		steps = append(steps, &types.PlanStep{
+			ID:                ksuid.New().String(),
+			Name:              "Extension Remove",
+			Type:              types.StepTypeECExtensionRemove,
+			Status:            types.StepStatusPending,
+			StatusDescription: "Pending extension removal",
+			Input: types.PlanStepInputECExtension{
+				Repos: newRepos,
+				Chart: chart,
+			},
+			Owner: types.StepOwnerECManager,
+		})
+	}
+
+	return steps, nil
+}
+
 func executeECExtensionAdd(p *types.Plan, step *types.PlanStep) error {
 	in, ok := step.Input.(types.PlanStepInputECExtension)
 	if !ok {
 		return errors.New("invalid input for embedded cluster extension add step")
 	}
-	if err := websocket.AddExtension(in.Repos, in.Chart, p.AppSlug, p.VersionLabel, step.ID); err != nil {
-		return errors.Wrap(err, "add extension")
+
+	if step.Status == types.StepStatusPending {
+		if err := websocket.AddExtension(in.Repos, in.Chart, p.AppSlug, p.VersionLabel, step.ID); err != nil {
+			return errors.Wrap(err, "add extension")
+		}
+	}
+
+	if err := waitForStep(p, step.ID); err != nil {
+		return errors.Wrap(err, "wait for embedded cluster extension add")
 	}
+
 	return nil
 }
 
@@ -34,9 +105,17 @@ func executeECExtensionUpgrade(p *types.Plan, step *types.PlanStep) error {
 	if !ok {
 		return errors.New("invalid input for embedded cluster extension upgrade step")
 	}
-	if err := websocket.UpgradeExtension(in.Repos, in.Chart, p.AppSlug, p.VersionLabel, step.ID); err != nil {
-		return errors.Wrap(err, "upgrade extension")
+
+	if step.Status == types.StepStatusPending {
+		if err := websocket.UpgradeExtension(in.Repos, in.Chart, p.AppSlug, p.VersionLabel, step.ID); err != nil {
+			return errors.Wrap(err, "upgrade extension")
+		}
+	}
+
+	if err := waitForStep(p, step.ID); err != nil {
+		return errors.Wrap(err, "wait for embedded cluster extension upgrade")
 	}
+
 	return nil
 }
 
@@ -45,9 +124,17 @@ func executeECExtensionRemove(p *types.Plan, step *types.PlanStep) error {
 	if !ok {
 		return errors.New("invalid input for embedded cluster extension remove step")
 	}
-	if err := websocket.RemoveExtension(in.Repos, in.Chart, p.AppSlug, p.VersionLabel, step.ID); err != nil {
-		return errors.Wrap(err, "remove extension")
+
+	if step.Status == types.StepStatusPending {
+		if err := websocket.RemoveExtension(in.Repos, in.Chart, p.AppSlug, p.VersionLabel, step.ID); err != nil {
+			return errors.Wrap(err, "remove extension")
+		}
 	}
+
+	if err := waitForStep(p, step.ID); err != nil {
+		return errors.Wrap(err, "wait for embedded cluster extension remove")
+	}
+
 	return nil
 }