Merge pull request #7567 from endocode/kbeecher/wip_etcd_migrate_keys

WIP: Add startup code to apiserver to migrate etcd keys

cmd/kube-apiserver/app/server.go

@@ -74,6 +74,7 @@ type APIServer struct {
 	EtcdServerList             util.StringList
 	EtcdConfigFile             string
 	EtcdPathPrefix             string
+	OldEtcdPathPrefix          string
 	CorsAllowedOriginList      util.StringList
 	AllowPrivileged            bool
 	PortalNet                  util.IPNet // TODO: make this a list
@@ -102,6 +103,7 @@ func NewAPIServer() *APIServer {
 		AuthorizationMode:      "AlwaysAllow",
 		AdmissionControl:       "AlwaysAdmit",
 		EtcdPathPrefix:         master.DefaultEtcdPathPrefix,
+		OldEtcdPathPrefix:      master.DefaultEtcdPathPrefix,
 		EnableLogsSupport:      true,
 		MasterServiceNamespace: api.NamespaceDefault,
 		ClusterName:            "kubernetes",
@@ -167,6 +169,7 @@ func (s *APIServer) AddFlags(fs *pflag.FlagSet) {
 	fs.Var(&s.EtcdServerList, "etcd-servers", "List of etcd servers to watch (http://ip:port), comma separated. Mutually exclusive with -etcd-config")
 	fs.StringVar(&s.EtcdConfigFile, "etcd-config", s.EtcdConfigFile, "The config file for the etcd client. Mutually exclusive with -etcd-servers.")
 	fs.StringVar(&s.EtcdPathPrefix, "etcd-prefix", s.EtcdPathPrefix, "The prefix for all resource paths in etcd.")
+	fs.StringVar(&s.OldEtcdPathPrefix, "old-etcd-prefix", s.OldEtcdPathPrefix, "The previous prefix for all resource paths in etcd, if any.")
 	fs.Var(&s.CorsAllowedOriginList, "cors-allowed-origins", "List of allowed origins for CORS, comma separated.  An allowed origin can be a regular expression to support subdomain matching.  If this list is empty CORS will not be enabled.")
 	fs.BoolVar(&s.AllowPrivileged, "allow-privileged", s.AllowPrivileged, "If true, allow privileged containers.")
 	fs.Var(&s.PortalNet, "portal-net", "A CIDR notation IP range from which to assign portal IPs. This must not overlap with any IP ranges assigned to nodes for pods.")
@@ -254,6 +257,14 @@ func (s *APIServer) Run(_ []string) error {
 		glog.Fatalf("Invalid storage version or misconfigured etcd: %v", err)
 	}
 
+	// TODO Is this the right place for migration to happen? Must *both* old and
+	// new etcd prefix params be supplied for this to be valid?
+	if s.OldEtcdPathPrefix != "" {
+		if err = helper.MigrateKeys(s.OldEtcdPathPrefix); err != nil {
+			glog.Fatalf("Migration of old etcd keys failed: %v", err)
+		}
+	}
+
 	n := net.IPNet(s.PortalNet)
 
 	authenticator, err := apiserver.NewAuthenticator(s.BasicAuthFile, s.ClientCAFile, s.TokenAuthFile)

pkg/tools/etcd_helper.go

@@ -545,6 +545,72 @@ func (h *EtcdHelper) PrefixEtcdKey(key string) string {
 	return path.Join("/", h.PathPrefix, key)
 }
 
+// Copies the key-value pairs from their old location to a new location based
+// on this helper's etcd prefix. All old keys without the prefix are then deleted.
+func (h *EtcdHelper) MigrateKeys(oldPathPrefix string) error {
+	// Check to see if a migration is necessary, i.e. is the oldPrefix different
+	// from the newPrefix?
+	if h.PathPrefix == oldPathPrefix {
+		return nil
+	}
+
+	// Get the root node
+	response, err := h.Client.Get(oldPathPrefix, false, true)
+	if err != nil {
+		glog.Infof("Couldn't get the existing etcd root node.")
+		return err
+	}
+
+	// Perform the migration
+	if err = h.migrateChildren(response.Node, oldPathPrefix); err != nil {
+		glog.Infof("Error performing the migration.")
+		return err
+	}
+
+	// Delete the old top-level entry recursively
+	// Quick sanity check: Did the process at least create a new top-level entry?
+	if _, err = h.Client.Get(h.PathPrefix, false, false); err != nil {
+		glog.Infof("Couldn't get the new etcd root node.")
+		return err
+	} else {
+		if _, err = h.Client.Delete(oldPathPrefix, true); err != nil {
+			glog.Infof("Couldn't delete the old etcd root node.")
+			return err
+		}
+	}
+	return nil
+}
+
+// This recurses through the etcd registry. Each key-value pair is copied with
+// to a new pair with a prefixed key.
+func (h *EtcdHelper) migrateChildren(parent *etcd.Node, oldPathPrefix string) error {
+	for _, child := range parent.Nodes {
+		if child.Dir && len(child.Nodes) > 0 {
+			// Descend into this directory
+			h.migrateChildren(child, oldPathPrefix)
+
+			// All children have been migrated, so this directory has
+			// already been automatically added.
+			continue
+		}
+
+		// Check if already prefixed (maybe we got interrupted in last attempt)
+		if strings.HasPrefix(child.Key, h.PathPrefix) {
+			// Skip this iteration
+			continue
+		}
+
+		// Create new entry
+		newKey := path.Join("/", h.PathPrefix, strings.TrimPrefix(child.Key, oldPathPrefix))
+		if _, err := h.Client.Create(newKey, child.Value, 0); err != nil {
+			// Assuming etcd is still available, this is due to the key
+			// already existing, in which case we can skip.
+			continue
+		}
+	}
+	return nil
+}
+
 // GetEtcdVersion performs a version check against the provided Etcd server,
 // returning the string response, and error (if any).
 func GetEtcdVersion(host string) (string, error) {

test/integration/etcd_tools_test.go

@@ -145,3 +145,47 @@ func TestWatch(t *testing.T) {
 		}
 	})
 }
+
+func TestMigrateKeys(t *testing.T) {
+	withEtcdKey(func(oldPrefix string) {
+		client := newEtcdClient()
+		helper := tools.NewEtcdHelper(client, testapi.Codec(), oldPrefix)
+
+		key1 := oldPrefix + "/obj1"
+		key2 := oldPrefix + "/foo/obj2"
+		key3 := oldPrefix + "/foo/bar/obj3"
+
+		// Create a new entres - these are the 'existing' entries with old prefix
+		_, _ = helper.Client.Create(key1, "foo", 0)
+		_, _ = helper.Client.Create(key2, "foo", 0)
+		_, _ = helper.Client.Create(key3, "foo", 0)
+
+		// Change the helper to a new prefix
+		newPrefix := "/kubernetes.io"
+		helper = tools.NewEtcdHelper(client, testapi.Codec(), newPrefix)
+
+		// Migrate the keys
+		err := helper.MigrateKeys(oldPrefix)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		// Check the resources are at the correct new location
+		newNames := []string{
+			newPrefix + "/obj1",
+			newPrefix + "/foo/obj2",
+			newPrefix + "/foo/bar/obj3",
+		}
+		for _, name := range newNames {
+			_, err := helper.Client.Get(name, false, false)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+		}
+
+		// Check the old locations are removed
+		if _, err := helper.Client.Get(oldPrefix, false, false); err == nil {
+			t.Fatalf("Old directory still exists.")
+		}
+	})
+}