feature(deisctl) allow for graceful in-place upgrades

Author: iancoffey

deisctl/backend/backend.go

@@ -12,6 +12,7 @@ type Backend interface {
 	Start([]string, *sync.WaitGroup, io.Writer, io.Writer)
 	Stop([]string, *sync.WaitGroup, io.Writer, io.Writer)
 	Scale(string, int, *sync.WaitGroup, io.Writer, io.Writer)
+	RollingRestart(string, *sync.WaitGroup, io.Writer, io.Writer)
 	SSH(string) error
 	SSHExec(string, string) error
 	ListUnits() error

deisctl/backend/fleet/rolling_restart.go

@@ -0,0 +1,37 @@
+package fleet
+
+import (
+	"fmt"
+	"io"
+	"sync"
+)
+
+// RollingRestart for instance units
+func (c *FleetClient) RollingRestart(component string, wg *sync.WaitGroup, out, ew io.Writer) {
+	if component != "router" {
+		fmt.Fprint(ew, "invalid component. supported for: router")
+		return
+	}
+
+	components, err := c.Units(component)
+	if err != nil {
+		io.WriteString(ew, err.Error())
+		return
+	}
+	if len(components) < 1 {
+		fmt.Fprint(ew, "rolling restart requires at least 1 component")
+		return
+	}
+	for num := range components {
+		unitName := fmt.Sprintf("%s@%v", component, num+1)
+
+		c.Stop([]string{unitName}, wg, out, ew)
+		wg.Wait()
+		c.Destroy([]string{unitName}, wg, out, ew)
+		wg.Wait()
+		c.Create([]string{unitName}, wg, out, ew)
+		wg.Wait()
+		c.Start([]string{unitName}, wg, out, ew)
+		wg.Wait()
+	}
+}

deisctl/client/client.go

@@ -28,6 +28,9 @@ type DeisCtlClient interface {
 	Status(argv []string) error
 	Stop(argv []string) error
 	Uninstall(argv []string) error
+	UpgradePrep(argv []string) error
+	UpgradeTakeover(argv []string) error
+	RollingRestart(argv []string) error
 }
 
 // Client uses a backend to implement the DeisCtlClient interface.
@@ -57,6 +60,49 @@ func NewClient(requestedBackend string) (*Client, error) {
 	return &Client{Backend: backend}, nil
 }
 
+// UpgradePrep prepares a running cluster to be upgraded
+func (c *Client) UpgradePrep(argv []string) error {
+	usage := `Prepare platform for graceful upgrade.
+
+Usage:
+  deisctl upgrade-prep [options]
+`
+	if _, err := docopt.Parse(usage, argv, true, "", false); err != nil {
+		return err
+	}
+
+	return cmd.UpgradePrep(c.Backend)
+}
+
+// UpgradeTakeover gracefully restarts a cluster prepared with upgrade-prep
+func (c *Client) UpgradeTakeover(argv []string) error {
+	usage := `Complete the upgrade of a prepped cluster.
+
+Usage:
+  deisctl upgrade-takeover [options]
+`
+	if _, err := docopt.Parse(usage, argv, true, "", false); err != nil {
+		return err
+	}
+
+	return cmd.UpgradeTakeover(c.Backend)
+}
+
+// RollingRestart attempts a rolling restart of an instance unit
+func (c *Client) RollingRestart(argv []string) error {
+	usage := `Perform a rolling restart of an instance unit.
+
+Usage:
+  deisctl rolling-restart <target>
+`
+	args, err := docopt.Parse(usage, argv, true, "", false)
+	if err != nil {
+		return err
+	}
+
+	return cmd.RollingRestart(args["<target>"].(string), c.Backend)
+}
+
 // Config gets or sets a configuration value from the cluster.
 //
 // A configuration value is stored and retrieved from a key/value store (in this case, etcd)

deisctl/cmd/cmd.go

@@ -97,6 +97,16 @@ func Start(targets []string, b backend.Backend) error {
 	return nil
 }
 
+// RollingRestart restart instance unit in a rolling manner
+func RollingRestart(target string, b backend.Backend) error {
+	var wg sync.WaitGroup
+
+	b.RollingRestart(target, &wg, Stdout, Stderr)
+	wg.Wait()
+
+	return nil
+}
+
 // CheckRequiredKeys exist in etcd
 func CheckRequiredKeys() error {
 	if err := config.CheckConfig("/deis/platform/", "domain"); err != nil {

deisctl/cmd/cmd_test.go

@@ -13,6 +13,8 @@ import (
 	"testing"
 
 	"github.com/deis/deis/deisctl/backend"
+	"github.com/deis/deis/deisctl/etcdclient"
+	"github.com/deis/deis/deisctl/test/mock"
 	"github.com/deis/deis/deisctl/units"
 )
 
@@ -21,6 +23,7 @@ type backendStub struct {
 	stoppedUnits     []string
 	installedUnits   []string
 	uninstalledUnits []string
+	restartedUnits   []string
 	expected         bool
 }
 
@@ -46,6 +49,10 @@ func (backend *backendStub) Scale(component string, num int, wg *sync.WaitGroup,
 		backend.expected = false
 	}
 }
+func (backend *backendStub) RollingRestart(target string, wg *sync.WaitGroup, out, ew io.Writer) {
+	backend.restartedUnits = append(backend.restartedUnits, target)
+}
+
 func (backend *backendStub) ListUnits() error {
 	return nil
 }
@@ -270,6 +277,57 @@ func TestStartSwarm(t *testing.T) {
 	}
 }
 
+func TestRollingRestart(t *testing.T) {
+	t.Parallel()
+
+	b := backendStub{}
+	expected := []string{"router"}
+
+	RollingRestart("router", &b)
+
+	if !reflect.DeepEqual(b.restartedUnits, expected) {
+		t.Error(fmt.Errorf("Expected %v, Got %v", expected, b.restartedUnits))
+	}
+}
+
+func TestUpgradePrep(t *testing.T) {
+	t.Parallel()
+
+	b := backendStub{}
+	expected := []string{"database", "registry@*", "controller", "builder", "logger", "logspout", "store-volume",
+		"store-gateway@*", "store-metadata", "store-daemon", "store-monitor"}
+
+	UpgradePrep(&b)
+
+	if !reflect.DeepEqual(b.stoppedUnits, expected) {
+		t.Error(fmt.Errorf("Expected %v, Got %v", expected, b.stoppedUnits))
+	}
+}
+
+func TestUpgradeTakeover(t *testing.T) {
+	t.Parallel()
+	testMock := mock.Client{Expected: []*etcdclient.ServiceKey{{Key: "/deis/services/app1", Value: "foo", TTL: 10},
+		{Key: "/deis/services/app2", Value: "8000", TTL: 10}}}
+
+	b := backendStub{}
+	expectedRestarted := []string{"router"}
+	expectedStarted := []string{"publisher", "store-monitor", "store-daemon", "store-metadata",
+		"store-gateway@*", "store-volume", "logger", "logspout", "database", "registry@*",
+		"controller", "builder", "publisher", "router@*", "database", "registry@*",
+		"controller", "builder", "publisher", "router@*"}
+
+	if err := doUpgradeTakeOver(&b, testMock); err != nil {
+		t.Error(fmt.Errorf("Takeover failed: %v", err))
+	}
+
+	if !reflect.DeepEqual(b.restartedUnits, expectedRestarted) {
+		t.Error(fmt.Errorf("Expected %v, Got %v", expectedRestarted, b.restartedUnits))
+	}
+	if !reflect.DeepEqual(b.startedUnits, expectedStarted) {
+		t.Error(fmt.Errorf("Expected %v, Got %v", expectedStarted, b.startedUnits))
+	}
+}
+
 func TestStop(t *testing.T) {
 	t.Parallel()
 

deisctl/cmd/upgrade.go

@@ -0,0 +1,110 @@
+package cmd
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/deis/deis/deisctl/backend"
+	"github.com/deis/deis/deisctl/etcdclient"
+)
+
+// UpgradePrep stops and uninstalls all components except router and publisher
+func UpgradePrep(b backend.Backend) error {
+	var wg sync.WaitGroup
+
+	b.Stop([]string{"database", "registry@*", "controller", "builder", "logger", "logspout"}, &wg, Stdout, Stderr)
+	wg.Wait()
+	b.Destroy([]string{"database", "registry@*", "controller", "builder", "logger", "logspout"}, &wg, Stdout, Stderr)
+	wg.Wait()
+
+	b.Stop([]string{"store-volume", "store-gateway@*"}, &wg, Stdout, Stderr)
+	wg.Wait()
+	b.Destroy([]string{"store-volume", "store-gateway@*"}, &wg, Stdout, Stderr)
+	wg.Wait()
+
+	b.Stop([]string{"store-metadata"}, &wg, Stdout, Stderr)
+	wg.Wait()
+	b.Destroy([]string{"store-metadata"}, &wg, Stdout, Stderr)
+	wg.Wait()
+
+	b.Stop([]string{"store-daemon"}, &wg, Stdout, Stderr)
+	wg.Wait()
+	b.Destroy([]string{"store-daemon"}, &wg, Stdout, Stderr)
+	wg.Wait()
+
+	b.Stop([]string{"store-monitor"}, &wg, Stdout, Stderr)
+	wg.Wait()
+	b.Destroy([]string{"store-monitor"}, &wg, Stdout, Stderr)
+	wg.Wait()
+
+	fmt.Fprintln(Stdout, "The platform has been stopped, but applications are still serving traffic as normal.")
+	fmt.Fprintln(Stdout, "Your cluster is now ready for upgrade. Install a new deisctl version and run `deisctl upgrade-takeover`.")
+	fmt.Fprintln(Stdout, "For more details, see: http://docs.deis.io/en/latest/managing_deis/upgrading-deis/#graceful-upgrade")
+	return nil
+}
+
+func listPublishedServices(etcdClient etcdclient.Client) ([]*etcdclient.ServiceKey, error) {
+	nodes, err := etcdClient.GetRecursive("deis/services")
+	if err != nil {
+		return nil, err
+	}
+
+	return nodes, nil
+}
+
+func republishServices(ttl uint64, nodes []*etcdclient.ServiceKey, etcdClient etcdclient.Client) error {
+	for _, node := range nodes {
+		_, err := etcdClient.Update(node.Key, node.Value, ttl)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// UpgradeTakeover gracefully starts a platform stopped with UpgradePrep
+func UpgradeTakeover(b backend.Backend) error {
+	etcdClient, err := etcdclient.GetEtcdClient()
+	if err != nil {
+		return err
+	}
+
+	if err := doUpgradeTakeOver(b, etcdClient); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func doUpgradeTakeOver(b backend.Backend, etcdClient etcdclient.Client) error {
+	var wg sync.WaitGroup
+
+	nodes, err := listPublishedServices(etcdClient)
+	if err != nil {
+		return err
+	}
+
+	b.Stop([]string{"publisher"}, &wg, Stdout, Stderr)
+	wg.Wait()
+	b.Destroy([]string{"publisher"}, &wg, Stdout, Stderr)
+	wg.Wait()
+
+	if err := republishServices(1800, nodes, etcdClient); err != nil {
+		return err
+	}
+
+	b.RollingRestart("router", &wg, Stdout, Stderr)
+	wg.Wait()
+	b.Create([]string{"publisher"}, &wg, Stdout, Stderr)
+	wg.Wait()
+	b.Start([]string{"publisher"}, &wg, Stdout, Stderr)
+	wg.Wait()
+
+	installDefaultServices(b, false, &wg, Stdout, Stderr) // @fixme: hax?
+	wg.Wait()
+
+	startDefaultServices(b, false, &wg, Stdout, Stderr) // @fixme: hax?
+	wg.Wait()
+	return nil
+}

deisctl/config/api.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"strings"
 
+	"github.com/deis/deis/deisctl/etcdclient"
 	"github.com/deis/deis/deisctl/utils"
 )
 
@@ -23,7 +24,7 @@ var b64Keys = []string{"/deis/platform/sshPrivateKey"}
 
 // Config runs the config subcommand
 func Config(target string, action string, key []string) error {
-	client, err := getEtcdClient()
+	client, err := etcdclient.GetEtcdClient()
 	if err != nil {
 		return err
 	}
@@ -35,7 +36,7 @@ func Config(target string, action string, key []string) error {
 // and returns an error if a value is not found
 func CheckConfig(root string, k string) error {
 
-	client, err := getEtcdClient()
+	client, err := etcdclient.GetEtcdClient()
 	if err != nil {
 		return err
 	}
@@ -48,7 +49,7 @@ func CheckConfig(root string, k string) error {
 	return nil
 }
 
-func doConfig(target string, action string, key []string, client Client, w io.Writer) error {
+func doConfig(target string, action string, key []string, client etcdclient.Client, w io.Writer) error {
 	rootPath := "/deis/" + target + "/"
 
 	var vals []string
@@ -73,7 +74,7 @@ func doConfig(target string, action string, key []string, client Client, w io.Wr
 	return nil
 }
 
-func doConfigSet(client Client, root string, kvs []string) ([]string, error) {
+func doConfigSet(client etcdclient.Client, root string, kvs []string) ([]string, error) {
 	var result []string
 
 	for _, kv := range kvs {
@@ -100,7 +101,7 @@ func doConfigSet(client Client, root string, kvs []string) ([]string, error) {
 	return result, nil
 }
 
-func doConfigGet(client Client, root string, keys []string) ([]string, error) {
+func doConfigGet(client etcdclient.Client, root string, keys []string) ([]string, error) {
 	var result []string
 	for _, k := range keys {
 		val, err := client.Get(root + k)
@@ -112,7 +113,7 @@ func doConfigGet(client Client, root string, keys []string) ([]string, error) {
 	return result, nil
 }
 
-func doConfigRm(client Client, root string, keys []string) ([]string, error) {
+func doConfigRm(client etcdclient.Client, root string, keys []string) ([]string, error) {
 	var result []string
 	for _, k := range keys {
 		err := client.Delete(root + k)