Merge pull request #383 from sgotti/stolonctl_removekeeper

stolonctl: add removekeeper command

cmd/stolonctl/removekeeper.go

@@ -0,0 +1,95 @@
+// Copyright 2017 Sorint.lab
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"github.com/sorintlab/stolon/pkg/cluster"
+	"github.com/spf13/cobra"
+)
+
+var removeKeeperCmd = &cobra.Command{
+	Use:   "removekeeper [keeper uid]",
+	Short: "Removes keeper from cluster data",
+	Run:   removeKeeper,
+}
+
+func init() {
+	cmdStolonCtl.AddCommand(removeKeeperCmd)
+}
+
+func removeKeeper(cmd *cobra.Command, args []string) {
+	if len(args) > 1 {
+		die("too many arguments")
+	}
+
+	if len(args) == 0 {
+		die("keeper uid required")
+	}
+
+	keeperID := args[0]
+
+	store, err := NewStore()
+	if err != nil {
+		die("cannot create store: %v", err)
+	}
+
+	cd, pair, err := getClusterData(store)
+	if err != nil {
+		die("cannot get cluster data: %v", err)
+	}
+	if cd.Cluster == nil {
+		die("no cluster spec available")
+	}
+	if cd.Cluster.Spec == nil {
+		die("no cluster spec available")
+	}
+
+	newCd := cd.DeepCopy()
+	keeperInfo := newCd.Keepers[keeperID]
+	if keeperInfo == nil {
+		die("keeper doesn't exist")
+	}
+
+	keeperDb := getDbForKeeper(newCd.DBs, keeperID)
+
+	if keeperDb != nil {
+		if newCd.Cluster.Status.Master == keeperDb.UID {
+			die("keeper assigned db is the current cluster master db.")
+		}
+	}
+
+	delete(newCd.Keepers, keeperID)
+	if keeperDb != nil {
+		delete(newCd.DBs, keeperDb.UID)
+	}
+
+	// NOTE: if the removed db is listed inside another db.Followers it'll will
+	// be cleaned by the sentinels
+
+	_, err = store.AtomicPutClusterData(newCd, pair)
+	if err != nil {
+		die("cannot update cluster data: %v", err)
+	}
+}
+
+func getDbForKeeper(dbs cluster.DBs, keeperID string) *cluster.DB {
+	for _, db := range dbs {
+		if db.Spec.KeeperUID == keeperID {
+			return db
+		}
+	}
+
+	return nil
+}

cmd/stolonctl/status.go

@@ -36,6 +36,12 @@ func init() {
 }
 
 func printTree(dbuid string, cd *cluster.ClusterData, level int, prefix string, tail bool) {
+	// skip not existing db: specified as a follower but not available in the
+	// clister spec (this should happen only when doing a stolonctl
+	// removekeeper)
+	if _, ok := cd.DBs[dbuid]; !ok {
+		return
+	}
 	out := prefix
 	if level > 0 {
 		if tail {

tests/integration/ha_test.go

@@ -1435,6 +1435,109 @@ func TestKeeperRemoval(t *testing.T) {
 	}
 }
 
+func testKeeperRemovalStolonCtl(t *testing.T, syncRepl bool) {
+	dir, err := ioutil.TempDir("", "stolon")
+	if err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+	defer os.RemoveAll(dir)
+
+	clusterName := uuid.NewV4().String()
+
+	initialClusterSpec := &cluster.ClusterSpec{
+		InitMode:               cluster.ClusterInitModeP(cluster.ClusterInitModeNew),
+		SleepInterval:          &cluster.Duration{Duration: 2 * time.Second},
+		FailInterval:           &cluster.Duration{Duration: 5 * time.Second},
+		ConvergenceTimeout:     &cluster.Duration{Duration: 30 * time.Second},
+		SynchronousReplication: cluster.BoolP(syncRepl),
+		MaxSynchronousStandbys: cluster.Uint16P(3),
+		PGParameters:           defaultPGParameters,
+	}
+
+	// Create 3 keepers
+	tks, tss, tp, tstore := setupServersCustom(t, clusterName, dir, 3, 1, initialClusterSpec)
+	defer shutdown(tks, tss, tp, tstore)
+
+	storeEndpoints := fmt.Sprintf("%s:%s", tstore.listenAddress, tstore.port)
+	storePath := filepath.Join(common.StoreBasePath, clusterName)
+	sm := store.NewStoreManager(tstore.store, storePath)
+
+	master, standbys := waitMasterStandbysReady(t, sm, tks)
+
+	if syncRepl {
+		if err := WaitClusterDataSynchronousStandbys([]string{standbys[0].uid, standbys[1].uid}, sm, 30*time.Second); err != nil {
+			t.Fatalf("expected synchronous standbys")
+		}
+	}
+
+	if err := populate(t, master); err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+	if err := write(t, master, 1, 1); err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+	c, err := getLines(t, master)
+	if err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+	if c != 1 {
+		t.Fatalf("wrong number of lines, want: %d, got: %d", 1, c)
+	}
+	if err := waitLines(t, standbys[0], 1, 30*time.Second); err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+
+	// remove master from the cluster data, must fail
+	err = StolonCtl(clusterName, tstore.storeBackend, storeEndpoints, "removekeeper", master.uid)
+	t.Logf("received err: %v", err)
+	if err == nil {
+		t.Fatalf("expected err")
+	}
+
+	// stop standbys[0]
+	t.Logf("Stopping standbys[0] keeper: %s", standbys[0].uid)
+	standbys[0].Stop()
+
+	// remove standby[0] from the cluster data
+	err = StolonCtl(clusterName, tstore.storeBackend, storeEndpoints, "removekeeper", standbys[0].uid)
+	if err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+
+	// get current stanbdys[0] db uid
+	cd, _, err := sm.GetClusterData()
+	if err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+
+	// check that the number of keepers and dbs is 1
+	if len(cd.Keepers) != 2 {
+		t.Fatalf("expected 2 keeper in cluster data, got: %d", len(cd.Keepers))
+	}
+	if len(cd.DBs) != 2 {
+		t.Fatalf("expected 2 db in cluster data, got: %d", len(cd.DBs))
+	}
+
+	// restart standbys[0]
+	t.Logf("Starting standbys[0] keeper: %s", standbys[0].uid)
+	standbys[0].Start()
+
+	// it should reenter the cluster with the same uid but a new assigned db uid
+	if err := waitLines(t, standbys[0], 1, 30*time.Second); err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+}
+
+func TestKeeperRemovalStolonCtl(t *testing.T) {
+	t.Parallel()
+	testKeeperRemovalStolonCtl(t, false)
+}
+
+func TestKeeperRemovalStolonCtlSyncRepl(t *testing.T) {
+	t.Parallel()
+	testKeeperRemovalStolonCtl(t, true)
+}
+
 func TestStandbyCantSync(t *testing.T) {
 	t.Parallel()