mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-07-23 19:56:01 +00:00
Merge pull request #7567 from endocode/kbeecher/wip_etcd_migrate_keys
WIP: Add startup code to apiserver to migrate etcd keys
This commit is contained in:
commit
5461231c1c
@ -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)
|
||||
|
@ -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) {
|
||||
|
@ -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.")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user