diff --git a/pkg/cli/etcdsnapshot/etcd_snapshot.go b/pkg/cli/etcdsnapshot/etcd_snapshot.go index bdbd3c3e9a..b2d8525230 100644 --- a/pkg/cli/etcdsnapshot/etcd_snapshot.go +++ b/pkg/cli/etcdsnapshot/etcd_snapshot.go @@ -64,7 +64,7 @@ func commandSetup(app *cli.Context, cfg *cmds.Server, sc *server.Config) error { sc.ControlConfig.Runtime.ETCDServerCA = filepath.Join(dataDir, "tls", "etcd", "server-ca.crt") sc.ControlConfig.Runtime.ClientETCDCert = filepath.Join(dataDir, "tls", "etcd", "client.crt") sc.ControlConfig.Runtime.ClientETCDKey = filepath.Join(dataDir, "tls", "etcd", "client.key") - sc.ControlConfig.Runtime.KubeConfigSupervisor = filepath.Join(dataDir, "cred", "supervisor.kubeconfig") + sc.ControlConfig.Runtime.KubeConfigAdmin = filepath.Join(dataDir, "cred", "admin.kubeconfig") return nil } @@ -98,11 +98,11 @@ func save(app *cli.Context, cfg *cmds.Server) error { ctx := signals.SetupSignalContext() e := etcd.NewETCD() - if err := e.SetControlConfig(ctx, &serverConfig.ControlConfig); err != nil { + if err := e.SetControlConfig(&serverConfig.ControlConfig); err != nil { return err } - initialized, err := e.IsInitialized(ctx, &serverConfig.ControlConfig) + initialized, err := e.IsInitialized() if err != nil { return err } @@ -116,7 +116,7 @@ func save(app *cli.Context, cfg *cmds.Server) error { return err } - sc, err := server.NewContext(ctx, serverConfig.ControlConfig.Runtime.KubeConfigSupervisor) + sc, err := server.NewContext(ctx, serverConfig.ControlConfig.Runtime.KubeConfigAdmin, false) if err != nil { return err } @@ -146,11 +146,11 @@ func delete(app *cli.Context, cfg *cmds.Server) error { ctx := signals.SetupSignalContext() e := etcd.NewETCD() - if err := e.SetControlConfig(ctx, &serverConfig.ControlConfig); err != nil { + if err := e.SetControlConfig(&serverConfig.ControlConfig); err != nil { return err } - sc, err := server.NewContext(ctx, serverConfig.ControlConfig.Runtime.KubeConfigSupervisor) + sc, err := server.NewContext(ctx, serverConfig.ControlConfig.Runtime.KubeConfigAdmin, false) if err != nil { return err } @@ -186,7 +186,7 @@ func list(app *cli.Context, cfg *cmds.Server) error { ctx := signals.SetupSignalContext() e := etcd.NewETCD() - if err := e.SetControlConfig(ctx, &serverConfig.ControlConfig); err != nil { + if err := e.SetControlConfig(&serverConfig.ControlConfig); err != nil { return err } @@ -252,11 +252,11 @@ func prune(app *cli.Context, cfg *cmds.Server) error { ctx := signals.SetupSignalContext() e := etcd.NewETCD() - if err := e.SetControlConfig(ctx, &serverConfig.ControlConfig); err != nil { + if err := e.SetControlConfig(&serverConfig.ControlConfig); err != nil { return err } - sc, err := server.NewContext(ctx, serverConfig.ControlConfig.Runtime.KubeConfigSupervisor) + sc, err := server.NewContext(ctx, serverConfig.ControlConfig.Runtime.KubeConfigAdmin, false) if err != nil { return err } diff --git a/pkg/cluster/bootstrap.go b/pkg/cluster/bootstrap.go index 592ca0f71a..a2c63a974e 100644 --- a/pkg/cluster/bootstrap.go +++ b/pkg/cluster/bootstrap.go @@ -82,7 +82,7 @@ func (c *Cluster) shouldBootstrapLoad(ctx context.Context) (bool, bool, error) { if c.managedDB != nil { c.config.Runtime.HTTPBootstrap = true - isInitialized, err := c.managedDB.IsInitialized(ctx, c.config) + isInitialized, err := c.managedDB.IsInitialized() if err != nil { return false, false, err } @@ -430,7 +430,7 @@ func (c *Cluster) Snapshot(ctx context.Context, config *config.Control) error { if c.managedDB == nil { return errors.New("unable to perform etcd snapshot on non-etcd system") } - return c.managedDB.Snapshot(ctx, config) + return c.managedDB.Snapshot(ctx) } // compareConfig verifies that the config of the joining control plane node coincides with the cluster's config @@ -503,7 +503,7 @@ func (c *Cluster) reconcileEtcd(ctx context.Context) error { }() e := etcd.NewETCD() - if err := e.SetControlConfig(reconcileCtx, c.config); err != nil { + if err := e.SetControlConfig(c.config); err != nil { return err } if err := e.StartEmbeddedTemporary(reconcileCtx); err != nil { diff --git a/pkg/cluster/https.go b/pkg/cluster/https.go index 71aafe180b..6ac3e6cb9e 100644 --- a/pkg/cluster/https.go +++ b/pkg/cluster/https.go @@ -15,7 +15,6 @@ import ( "github.com/gorilla/mux" "github.com/k3s-io/k3s/pkg/daemons/config" - "github.com/k3s-io/k3s/pkg/etcd" "github.com/k3s-io/k3s/pkg/version" "github.com/rancher/dynamiclistener" "github.com/rancher/dynamiclistener/factory" @@ -33,9 +32,14 @@ import ( // and will sync the certs into the Kubernetes datastore, with a local disk cache. func (c *Cluster) newListener(ctx context.Context) (net.Listener, http.Handler, error) { if c.managedDB != nil { - if _, err := os.Stat(etcd.ResetFile(c.config)); err == nil { - // delete the dynamic listener file if it exists after restoration to fix restoration - // on fresh nodes + resetDone, err := c.managedDB.IsReset() + if err != nil { + return nil, nil, err + } + if resetDone { + // delete the dynamic listener TLS secret cache after restoring, + // to ensure that dynamiclistener doesn't sync the old secret over the top + // of whatever was just restored. os.Remove(filepath.Join(c.config.DataDir, "tls/dynamic-cert.json")) } } @@ -104,8 +108,8 @@ func (c *Cluster) initClusterAndHTTPS(ctx context.Context) error { return err } - // Config the cluster database and allow it to add additional request handlers - handler, err = c.initClusterDB(ctx, handler) + // Register database request handlers and controller callbacks + handler, err = c.registerDBHandlers(handler) if err != nil { return err } diff --git a/pkg/cluster/managed.go b/pkg/cluster/managed.go index 4d13bb7534..e88ed79512 100644 --- a/pkg/cluster/managed.go +++ b/pkg/cluster/managed.go @@ -9,14 +9,12 @@ import ( "net/http" "net/url" "os" - "strings" "time" "github.com/k3s-io/k3s/pkg/cluster/managed" "github.com/k3s-io/k3s/pkg/etcd" "github.com/k3s-io/k3s/pkg/nodepassword" "github.com/k3s-io/k3s/pkg/version" - "github.com/k3s-io/kine/pkg/endpoint" "github.com/sirupsen/logrus" apierrors "k8s.io/apimachinery/pkg/api/errors" ) @@ -57,63 +55,62 @@ func (c *Cluster) start(ctx context.Context) error { if c.managedDB == nil { return nil } - resetFile := etcd.ResetFile(c.config) rebootstrap := func() error { return c.storageBootstrap(ctx) } + resetDone, err := c.managedDB.IsReset() + if err != nil { + return err + } + if c.config.ClusterReset { // If we're restoring from a snapshot, don't check the reset-flag - just reset and restore. if c.config.ClusterResetRestorePath != "" { return c.managedDB.Reset(ctx, rebootstrap) } + // If the reset-flag doesn't exist, reset. This will create the reset-flag if it succeeds. - if _, err := os.Stat(resetFile); err != nil { - if !os.IsNotExist(err) { - return err - } + if !resetDone { return c.managedDB.Reset(ctx, rebootstrap) } + // The reset-flag exists, ask the user to remove it if they want to reset again. - return fmt.Errorf("Managed etcd cluster membership was previously reset, please remove the cluster-reset flag and start %s normally. If you need to perform another cluster reset, you must first manually delete the %s file", version.Program, resetFile) + return fmt.Errorf("Managed etcd cluster membership was previously reset, please remove the cluster-reset flag and start %s normally. "+ + "If you need to perform another cluster reset, you must first manually delete the file at %s", version.Program, c.managedDB.ResetFile()) } - // The reset-flag exists but we're not resetting; remove it - if _, err := os.Stat(resetFile); err == nil { - // Before removing reset file we need to delete the node passwd secret in case the node + if resetDone { + // If the cluster was reset, we need to delete the node passwd secret in case the node // password from the previously restored snapshot differs from the current password on disk. c.config.Runtime.ClusterControllerStarts["node-password-secret-cleanup"] = c.deleteNodePasswdSecret - os.Remove(resetFile) } + // Starting the managed database will clear the reset-flag if set return c.managedDB.Start(ctx, c.clientAccessInfo) } -// initClusterDB registers routes for database info with the http request handler -func (c *Cluster) initClusterDB(ctx context.Context, handler http.Handler) (http.Handler, error) { +// registerDBHandlers registers routes for database info with the http request handler +func (c *Cluster) registerDBHandlers(handler http.Handler) (http.Handler, error) { if c.managedDB == nil { return handler, nil } - if !strings.HasPrefix(c.config.Datastore.Endpoint, c.managedDB.EndpointName()+"://") { - c.config.Datastore = endpoint.Config{ - Endpoint: c.managedDB.EndpointName(), - } - } - - return c.managedDB.Register(ctx, c.config, handler) + return c.managedDB.Register(handler) } // assignManagedDriver assigns a driver based on a number of different configuration variables. // If a driver has been initialized it is used. -// If the configured endpoint matches the name of a driver, that driver is used. // If no specific endpoint has been requested and creating or joining has been requested, // we use the default driver. // If none of the above are true, no managed driver is assigned. func (c *Cluster) assignManagedDriver(ctx context.Context) error { // Check all managed drivers for an initialized database on disk; use one if found for _, driver := range managed.Registered() { - if ok, err := driver.IsInitialized(ctx, c.config); err != nil { + if err := driver.SetControlConfig(c.config); err != nil { + return err + } + if ok, err := driver.IsInitialized(); err != nil { return err } else if ok { c.managedDB = driver @@ -121,24 +118,9 @@ func (c *Cluster) assignManagedDriver(ctx context.Context) error { } } - // This is needed to allow downstreams to override driver selection logic by - // setting ServerConfig.Datastore.Endpoint such that it will match a driver's EndpointName - endpointType := strings.SplitN(c.config.Datastore.Endpoint, ":", 2)[0] - for _, driver := range managed.Registered() { - if endpointType == driver.EndpointName() { - c.managedDB = driver - return nil - } - } - // If we have been asked to initialize or join a cluster, do so using the default managed database. if c.config.Datastore.Endpoint == "" && (c.config.ClusterInit || (c.config.Token != "" && c.config.JoinURL != "")) { - for _, driver := range managed.Registered() { - if driver.EndpointName() == managed.Default() { - c.managedDB = driver - return nil - } - } + c.managedDB = managed.Default() } return nil diff --git a/pkg/cluster/managed/drivers.go b/pkg/cluster/managed/drivers.go index 5e844e0975..83f54f7519 100644 --- a/pkg/cluster/managed/drivers.go +++ b/pkg/cluster/managed/drivers.go @@ -9,19 +9,21 @@ import ( ) var ( - defaultDriver string - drivers []Driver + drivers []Driver ) type Driver interface { - IsInitialized(ctx context.Context, config *config.Control) (bool, error) - Register(ctx context.Context, config *config.Control, handler http.Handler) (http.Handler, error) + SetControlConfig(config *config.Control) error + IsInitialized() (bool, error) + Register(handler http.Handler) (http.Handler, error) Reset(ctx context.Context, reboostrap func() error) error + IsReset() (bool, error) + ResetFile() string Start(ctx context.Context, clientAccessInfo *clientaccess.Info) error Test(ctx context.Context) error Restore(ctx context.Context) error EndpointName() string - Snapshot(ctx context.Context, config *config.Control) error + Snapshot(ctx context.Context) error ReconcileSnapshotData(ctx context.Context) error GetMembersClientURLs(ctx context.Context) ([]string, error) RemoveSelf(ctx context.Context) error @@ -35,9 +37,6 @@ func Registered() []Driver { return drivers } -func Default() string { - if defaultDriver == "" && len(drivers) == 1 { - return drivers[0].EndpointName() - } - return defaultDriver +func Default() Driver { + return drivers[0] } diff --git a/pkg/etcd/etcd.go b/pkg/etcd/etcd.go index fbeb027e04..9fc913a8c6 100644 --- a/pkg/etcd/etcd.go +++ b/pkg/etcd/etcd.go @@ -26,6 +26,7 @@ import ( "github.com/google/uuid" "github.com/gorilla/mux" "github.com/k3s-io/k3s/pkg/clientaccess" + "github.com/k3s-io/k3s/pkg/cluster/managed" "github.com/k3s-io/k3s/pkg/daemons/config" "github.com/k3s-io/k3s/pkg/daemons/control/deps" "github.com/k3s-io/k3s/pkg/daemons/executor" @@ -107,6 +108,9 @@ var ( type NodeControllerGetter func() controllerv1.NodeController +// explicit interface check +var _ managed.Driver = &ETCD{} + type ETCD struct { client *clientv3.Client config *config.Control @@ -163,22 +167,16 @@ func (e *ETCD) EndpointName() string { return "etcd" } -// SetControlConfig sets the given config on the etcd struct. -func (e *ETCD) SetControlConfig(ctx context.Context, config *config.Control) error { +// SetControlConfig passes the cluster config into the etcd datastore. This is necessary +// because the config may not yet be fully built at the time the Driver instance is registered. +func (e *ETCD) SetControlConfig(config *config.Control) error { + if e.config != nil { + return errors.New("control config already set") + } + e.config = config - client, err := GetClient(ctx, e.config) - if err != nil { - return err - } - e.client = client - - go func() { - <-ctx.Done() - e.client.Close() - }() - - address, err := getAdvertiseAddress(config.PrivateIP) + address, err := getAdvertiseAddress(e.config.PrivateIP) if err != nil { return err } @@ -192,6 +190,13 @@ func (e *ETCD) SetControlConfig(ctx context.Context, config *config.Control) err // If it is still a learner or not a part of the cluster, an error is raised. // If it cannot be defragmented or has any alarms that cannot be disarmed, an error is raised. func (e *ETCD) Test(ctx context.Context) error { + if e.config == nil { + return errors.New("control config not set") + } + if e.client == nil { + return errors.New("etcd datastore is not started") + } + ctx, cancel := context.WithTimeout(ctx, testTimeout) defer cancel() @@ -242,14 +247,14 @@ func (e *ETCD) Test(ctx context.Context) error { return &MembershipError{Members: memberNameUrls, Self: e.name + "=" + e.peerURL()} } -// DBDir returns the path to dataDir/db/etcd -func DBDir(config *config.Control) string { +// dbDir returns the path to dataDir/db/etcd +func dbDir(config *config.Control) string { return filepath.Join(config.DataDir, "db", "etcd") } // walDir returns the path to etcdDBDir/member/wal func walDir(config *config.Control) string { - return filepath.Join(DBDir(config), "member", "wal") + return filepath.Join(dbDir(config), "member", "wal") } func sqliteFile(config *config.Control) string { @@ -258,18 +263,48 @@ func sqliteFile(config *config.Control) string { // nameFile returns the path to etcdDBDir/name. func nameFile(config *config.Control) string { - return filepath.Join(DBDir(config), "name") + return filepath.Join(dbDir(config), "name") +} + +// clearReset removes the reset file +func (e *ETCD) clearReset() error { + if err := os.Remove(e.ResetFile()); err != nil && !os.IsNotExist(err) { + return err + } + return nil +} + +// IsReset checks to see if the reset file exists, indicating that a cluster-reset has been completed successfully. +func (e *ETCD) IsReset() (bool, error) { + if e.config == nil { + return false, errors.New("control config not set") + } + + if _, err := os.Stat(e.ResetFile()); err != nil { + if !os.IsNotExist(err) { + return false, err + } + return false, nil + } + return true, nil } // ResetFile returns the path to etcdDBDir/reset-flag. -func ResetFile(config *config.Control) string { - return filepath.Join(config.DataDir, "db", "reset-flag") +func (e *ETCD) ResetFile() string { + if e.config == nil { + panic("control config not set") + } + return filepath.Join(e.config.DataDir, "db", "reset-flag") } // IsInitialized checks to see if a WAL directory exists. If so, we assume that etcd // has already been brought up at least once. -func (e *ETCD) IsInitialized(ctx context.Context, config *config.Control) (bool, error) { - dir := walDir(config) +func (e *ETCD) IsInitialized() (bool, error) { + if e.config == nil { + return false, errors.New("control config not set") + } + + dir := walDir(e.config) if s, err := os.Stat(dir); err == nil && s.IsDir() { return true, nil } else if os.IsNotExist(err) { @@ -287,12 +322,13 @@ func (e *ETCD) Reset(ctx context.Context, rebootstrap func() error) error { t := time.NewTicker(5 * time.Second) defer t.Stop() for range t.C { - // resetting the apiaddresses to nil since we are doing a restoration - if _, err := e.client.Put(ctx, AddressKey, ""); err != nil { - logrus.Warnf("failed to reset api addresses key in etcd: %v", err) - continue - } if err := e.Test(ctx); err == nil { + // reset the apiaddresses to nil since we are doing a restoration + if _, err := e.client.Put(ctx, AddressKey, ""); err != nil { + logrus.Warnf("failed to reset api addresses key in etcd: %v", err) + continue + } + members, err := e.client.MemberList(ctx) if err != nil { continue @@ -338,6 +374,10 @@ func (e *ETCD) Reset(ctx context.Context, rebootstrap func() error) error { } }() + if err := e.startClient(ctx); err != nil { + return err + } + // If asked to restore from a snapshot, do so if e.config.ClusterResetRestorePath != "" { if e.config.EtcdS3 { @@ -367,7 +407,7 @@ func (e *ETCD) Reset(ctx context.Context, rebootstrap func() error) error { return err } // touch a file to avoid multiple resets - if err := os.WriteFile(ResetFile(e.config), []byte{}, 0600); err != nil { + if err := os.WriteFile(e.ResetFile(), []byte{}, 0600); err != nil { return err } return e.newCluster(ctx, true) @@ -375,9 +415,13 @@ func (e *ETCD) Reset(ctx context.Context, rebootstrap func() error) error { // Start starts the datastore func (e *ETCD) Start(ctx context.Context, clientAccessInfo *clientaccess.Info) error { - isInitialized, err := e.IsInitialized(ctx, e.config) + isInitialized, err := e.IsInitialized() if err != nil { - return errors.Wrapf(err, "configuration validation failed") + return errors.Wrapf(err, "failed to check for initialized etcd datastore") + } + + if err := e.startClient(ctx); err != nil { + return err } if !e.config.EtcdDisableSnapshots { @@ -389,7 +433,7 @@ func (e *ETCD) Start(ctx context.Context, clientAccessInfo *clientaccess.Info) e if isInitialized { //check etcd dir permission - etcdDir := DBDir(e.config) + etcdDir := dbDir(e.config) info, err := os.Stat(etcdDir) if err != nil { return err @@ -440,6 +484,35 @@ func (e *ETCD) Start(ctx context.Context, clientAccessInfo *clientaccess.Info) e return nil } +// startClient sets up the config's datastore endpoints, and starts an etcd client connected to the server endpoint. +// The client is destroyed when the context is closed. +func (e *ETCD) startClient(ctx context.Context) error { + if e.client != nil { + return errors.New("etcd datastore already started") + } + + endpoints := getEndpoints(e.config) + e.config.Datastore.Endpoint = endpoints[0] + e.config.Datastore.BackendTLSConfig.CAFile = e.config.Runtime.ETCDServerCA + e.config.Datastore.BackendTLSConfig.CertFile = e.config.Runtime.ClientETCDCert + e.config.Datastore.BackendTLSConfig.KeyFile = e.config.Runtime.ClientETCDKey + + client, err := getClient(ctx, e.config, endpoints...) + if err != nil { + return err + } + e.client = client + + go func() { + <-ctx.Done() + client := e.client + e.client = nil + client.Close() + }() + + return nil +} + // join attempts to add a member to an existing cluster func (e *ETCD) join(ctx context.Context, clientAccessInfo *clientaccess.Info) error { clientCtx, cancel := context.WithTimeout(ctx, 20*time.Second) @@ -455,7 +528,7 @@ func (e *ETCD) join(ctx context.Context, clientAccessInfo *clientaccess.Info) er return err } - client, err := GetClient(clientCtx, e.config, clientURLs...) + client, err := getClient(clientCtx, e.config, clientURLs...) if err != nil { return err } @@ -516,33 +589,8 @@ func (e *ETCD) join(ctx context.Context, clientAccessInfo *clientaccess.Info) er }) } -// Register configures a new etcd client and adds db info routes for the http request handler. -func (e *ETCD) Register(ctx context.Context, config *config.Control, handler http.Handler) (http.Handler, error) { - e.config = config - - client, err := GetClient(ctx, e.config) - if err != nil { - return nil, err - } - e.client = client - - go func() { - <-ctx.Done() - e.client.Close() - }() - - address, err := getAdvertiseAddress(config.PrivateIP) - if err != nil { - return nil, err - } - e.address = address - - endpoints := getEndpoints(config) - e.config.Datastore.Endpoint = endpoints[0] - e.config.Datastore.BackendTLSConfig.CAFile = e.config.Runtime.ETCDServerCA - e.config.Datastore.BackendTLSConfig.CertFile = e.config.Runtime.ClientETCDCert - e.config.Datastore.BackendTLSConfig.KeyFile = e.config.Runtime.ClientETCDKey - +// Register adds db info routes for the http request handler, and registers cluster controller callbacks +func (e *ETCD) Register(handler http.Handler) (http.Handler, error) { e.config.Runtime.ClusterControllerStarts["etcd-node-metadata"] = func(ctx context.Context) { registerMetadataHandlers(ctx, e) } @@ -560,10 +608,10 @@ func (e *ETCD) Register(ctx context.Context, config *config.Control, handler htt // Tombstone file checking is unnecessary if we're not running etcd. if !e.config.DisableETCD { - tombstoneFile := filepath.Join(DBDir(e.config), "tombstone") + tombstoneFile := filepath.Join(dbDir(e.config), "tombstone") if _, err := os.Stat(tombstoneFile); err == nil { logrus.Infof("tombstone file has been detected, removing data dir to rejoin the cluster") - if _, err := backupDirWithRetention(DBDir(e.config), maxBackupRetention); err != nil { + if _, err := backupDirWithRetention(dbDir(e.config), maxBackupRetention); err != nil { return nil, err } } @@ -631,12 +679,12 @@ func (e *ETCD) infoHandler() http.Handler { }) } -// GetClient returns an etcd client connected to the specified endpoints. +// getClient returns an etcd client connected to the specified endpoints. // If no endpoints are provided, endpoints are retrieved from the provided runtime config. // If the runtime config does not list any endpoints, the default endpoint is used. // The returned client should be closed when no longer needed, in order to avoid leaking GRPC // client goroutines. -func GetClient(ctx context.Context, control *config.Control, endpoints ...string) (*clientv3.Client, error) { +func getClient(ctx context.Context, control *config.Control, endpoints ...string) (*clientv3.Client, error) { cfg, err := getClientConfig(ctx, control, endpoints...) if err != nil { return nil, err @@ -659,7 +707,6 @@ func getClientConfig(ctx context.Context, control *config.Control, endpoints ... DialTimeout: defaultDialTimeout, DialKeepAliveTime: defaultKeepAliveTime, DialKeepAliveTimeout: defaultKeepAliveTimeout, - AutoSyncInterval: defaultKeepAliveTimeout, PermitWithoutStream: true, } @@ -761,7 +808,7 @@ func (e *ETCD) migrateFromSQLite(ctx context.Context) error { } defer sqliteClient.Close() - etcdClient, err := GetClient(ctx, e.config) + etcdClient, err := getClient(ctx, e.config) if err != nil { return err } @@ -845,7 +892,7 @@ func (e *ETCD) cluster(ctx context.Context, reset bool, options executor.Initial ListenMetricsURLs: e.listenMetricsURLs(reset), ListenPeerURLs: e.listenPeerURLs(reset), AdvertiseClientURLs: e.advertiseClientURLs(reset), - DataDir: DBDir(e.config), + DataDir: dbDir(e.config), ServerTrust: executor.ServerTrust{ CertFile: e.config.Runtime.ServerETCDCert, KeyFile: e.config.Runtime.ServerETCDKey, @@ -868,7 +915,7 @@ func (e *ETCD) cluster(ctx context.Context, reset bool, options executor.Initial } func (e *ETCD) StartEmbeddedTemporary(ctx context.Context) error { - etcdDataDir := DBDir(e.config) + etcdDataDir := dbDir(e.config) tmpDataDir := etcdDataDir + "-tmp" os.RemoveAll(tmpDataDir) @@ -879,6 +926,23 @@ func (e *ETCD) StartEmbeddedTemporary(ctx context.Context) error { } }() + if e.client != nil { + return errors.New("etcd datastore already started") + } + + client, err := getClient(ctx, e.config) + if err != nil { + return err + } + e.client = client + + go func() { + <-ctx.Done() + client := e.client + e.client = nil + client.Close() + }() + if err := cp.Copy(etcdDataDir, tmpDataDir, cp.Options{PreserveOwner: true}); err != nil { return err } @@ -1198,25 +1262,10 @@ func snapshotDir(config *config.Control, create bool) (string, error) { // preSnapshotSetup checks to see if the necessary components are in place // to perform an Etcd snapshot. This is necessary primarily for on-demand // snapshots since they're performed before normal Etcd setup is completed. -func (e *ETCD) preSnapshotSetup(ctx context.Context, config *config.Control) error { +func (e *ETCD) preSnapshotSetup(ctx context.Context) error { if e.snapshotSem == nil { e.snapshotSem = semaphore.NewWeighted(maxConcurrentSnapshots) } - if e.client == nil { - if e.config == nil { - e.config = config - } - client, err := GetClient(ctx, e.config) - if err != nil { - return err - } - e.client = client - - go func() { - <-ctx.Done() - e.client.Close() - }() - } return nil } @@ -1308,8 +1357,8 @@ func (e *ETCD) decompressSnapshot(snapshotDir, snapshotFile string) (string, err // Snapshot attempts to save a new snapshot to the configured directory, and then clean up any old and failed // snapshots in excess of the retention limits. This method is used in the internal cron snapshot // system as well as used to do on-demand snapshots. -func (e *ETCD) Snapshot(ctx context.Context, config *config.Control) error { - if err := e.preSnapshotSetup(ctx, config); err != nil { +func (e *ETCD) Snapshot(ctx context.Context) error { + if err := e.preSnapshotSetup(ctx); err != nil { return err } if !e.snapshotSem.TryAcquire(maxConcurrentSnapshots) { @@ -1337,7 +1386,22 @@ func (e *ETCD) Snapshot(ctx context.Context, config *config.Control) error { } endpoints := getEndpoints(e.config) - status, err := e.client.Status(ctx, endpoints[0]) + var client *clientv3.Client + var err error + + // Use the internal client if possible, or create a new one + // if run from the CLI. + if e.client != nil { + client = e.client + } else { + client, err = getClient(ctx, e.config, endpoints...) + if err != nil { + return err + } + defer client.Close() + } + + status, err := client.Status(ctx, endpoints[0]) if err != nil { return errors.Wrap(err, "failed to check etcd status for snapshot") } @@ -1964,7 +2028,7 @@ func (e *ETCD) setSnapshotFunction(ctx context.Context) { // having all the nodes take a snapshot at the exact same time can lead to excessive retry thrashing // when updating the snapshot list configmap. time.Sleep(time.Duration(rand.Float64() * float64(snapshotJitterMax))) - if err := e.Snapshot(ctx, e.config); err != nil { + if err := e.Snapshot(ctx); err != nil { logrus.Error(err) } }))) @@ -1975,7 +2039,7 @@ func (e *ETCD) setSnapshotFunction(ctx context.Context) { // completion. func (e *ETCD) Restore(ctx context.Context) error { // check the old etcd data dir - oldDataDir := DBDir(e.config) + "-old-" + strconv.Itoa(int(time.Now().Unix())) + oldDataDir := dbDir(e.config) + "-old-" + strconv.Itoa(int(time.Now().Unix())) if e.config.ClusterResetRestorePath == "" { return errors.New("no etcd restore path was specified") } @@ -2002,7 +2066,7 @@ func (e *ETCD) Restore(ctx context.Context) error { } // move the data directory to a temp path - if err := os.Rename(DBDir(e.config), oldDataDir); err != nil { + if err := os.Rename(dbDir(e.config), oldDataDir); err != nil { return err } @@ -2016,7 +2080,7 @@ func (e *ETCD) Restore(ctx context.Context) error { return snapshot.NewV3(lg).Restore(snapshot.RestoreConfig{ SnapshotPath: restorePath, Name: e.name, - OutputDataDir: DBDir(e.config), + OutputDataDir: dbDir(e.config), OutputWALDir: walDir(e.config), PeerURLs: []string{e.peerURL()}, InitialCluster: e.name + "=" + e.peerURL(), @@ -2112,7 +2176,7 @@ func backupDirWithRetention(dir string, maxBackupRetention int) (string, error) // GetAPIServerURLsFromETCD will try to fetch the version.Program/apiaddresses key from etcd // and unmarshal it to a list of apiserver endpoints. func GetAPIServerURLsFromETCD(ctx context.Context, cfg *config.Control) ([]string, error) { - cl, err := GetClient(ctx, cfg) + cl, err := getClient(ctx, cfg) if err != nil { return nil, err } @@ -2166,8 +2230,8 @@ func (e *ETCD) RemoveSelf(ctx context.Context) error { } // backup the data dir to avoid issues when re-enabling etcd - oldDataDir := DBDir(e.config) + "-old-" + strconv.Itoa(int(time.Now().Unix())) + oldDataDir := dbDir(e.config) + "-old-" + strconv.Itoa(int(time.Now().Unix())) // move the data directory to a temp path - return os.Rename(DBDir(e.config), oldDataDir) + return os.Rename(dbDir(e.config), oldDataDir) } diff --git a/pkg/etcd/etcd_test.go b/pkg/etcd/etcd_test.go index f71ad1e47b..c3af06284a 100644 --- a/pkg/etcd/etcd_test.go +++ b/pkg/etcd/etcd_test.go @@ -118,7 +118,11 @@ func Test_UnitETCD_IsInitialized(t *testing.T) { t.Errorf("Prep for ETCD.IsInitialized() failed = %v", err) return } - got, err := e.IsInitialized(tt.args.ctx, tt.args.config) + if err := e.SetControlConfig(tt.args.config); err != nil { + t.Errorf("ETCD.SetControlConfig() failed= %v", err) + return + } + got, err := e.IsInitialized() if (err != nil) != tt.wantErr { t.Errorf("ETCD.IsInitialized() error = %v, wantErr %v", err, tt.wantErr) return @@ -170,17 +174,17 @@ func Test_UnitETCD_Register(t *testing.T) { if err := testutil.GenerateRuntime(cnf); err != nil { return err } - if err := os.MkdirAll(DBDir(cnf), 0700); err != nil { + if err := os.MkdirAll(dbDir(cnf), 0700); err != nil { return err } - tombstoneFile := filepath.Join(DBDir(cnf), "tombstone") + tombstoneFile := filepath.Join(dbDir(cnf), "tombstone") if _, err := os.Create(tombstoneFile); err != nil { return err } return nil }, teardown: func(cnf *config.Control) error { - tombstoneFile := filepath.Join(DBDir(cnf), "tombstone") + tombstoneFile := filepath.Join(dbDir(cnf), "tombstone") os.Remove(tombstoneFile) testutil.CleanupDataDir(cnf) return nil @@ -196,7 +200,11 @@ func Test_UnitETCD_Register(t *testing.T) { t.Errorf("Setup for ETCD.Register() failed = %v", err) return } - _, err := e.Register(tt.args.ctx, tt.args.config, tt.args.handler) + if err := e.SetControlConfig(tt.args.config); err != nil { + t.Errorf("ETCD.SetControlConfig() failed = %v", err) + return + } + _, err := e.Register(tt.args.handler) if (err != nil) != tt.wantErr { t.Errorf("ETCD.Register() error = %v, wantErr %v", err, tt.wantErr) return @@ -244,17 +252,13 @@ func Test_UnitETCD_Start(t *testing.T) { ctxInfo.ctx, ctxInfo.cancel = context.WithCancel(context.Background()) e.config.EtcdDisableSnapshots = true testutil.GenerateRuntime(e.config) - client, err := GetClient(ctxInfo.ctx, e.config) - e.client = client - - return err + return nil }, teardown: func(e *ETCD, ctxInfo *contextInfo) error { // RemoveSelf will fail with a specific error, but it still does cleanup for testing purposes if err := e.RemoveSelf(ctxInfo.ctx); err != nil && err.Error() != etcdserver.ErrNotEnoughStartedMembers.Error() { return err } - e.client.Close() ctxInfo.cancel() time.Sleep(10 * time.Second) testutil.CleanupDataDir(e.config) @@ -275,17 +279,13 @@ func Test_UnitETCD_Start(t *testing.T) { setup: func(e *ETCD, ctxInfo *contextInfo) error { ctxInfo.ctx, ctxInfo.cancel = context.WithCancel(context.Background()) testutil.GenerateRuntime(e.config) - client, err := GetClient(ctxInfo.ctx, e.config) - e.client = client - - return err + return nil }, teardown: func(e *ETCD, ctxInfo *contextInfo) error { // RemoveSelf will fail with a specific error, but it still does cleanup for testing purposes if err := e.RemoveSelf(ctxInfo.ctx); err != nil && err.Error() != etcdserver.ErrNotEnoughStartedMembers.Error() { return err } - e.client.Close() ctxInfo.cancel() time.Sleep(5 * time.Second) testutil.CleanupDataDir(e.config) @@ -308,11 +308,6 @@ func Test_UnitETCD_Start(t *testing.T) { if err := testutil.GenerateRuntime(e.config); err != nil { return err } - client, err := GetClient(ctxInfo.ctx, e.config) - if err != nil { - return err - } - e.client = client return os.MkdirAll(walDir(e.config), 0700) }, teardown: func(e *ETCD, ctxInfo *contextInfo) error { @@ -320,7 +315,6 @@ func Test_UnitETCD_Start(t *testing.T) { if err := e.RemoveSelf(ctxInfo.ctx); err != nil && err.Error() != etcdserver.ErrNotEnoughStartedMembers.Error() { return err } - e.client.Close() ctxInfo.cancel() time.Sleep(5 * time.Second) testutil.CleanupDataDir(e.config) diff --git a/pkg/etcd/member_controller.go b/pkg/etcd/member_controller.go index c38fa174e4..ebf30ffb65 100644 --- a/pkg/etcd/member_controller.go +++ b/pkg/etcd/member_controller.go @@ -7,8 +7,10 @@ import ( "github.com/k3s-io/k3s/pkg/util" "github.com/k3s-io/k3s/pkg/version" + "github.com/pkg/errors" controllerv1 "github.com/rancher/wrangler/pkg/generated/controllers/core/v1" "github.com/sirupsen/logrus" + "go.etcd.io/etcd/api/v3/v3rpc/rpctypes" v1 "k8s.io/api/core/v1" ) @@ -49,52 +51,58 @@ func (e *etcdMemberHandler) sync(key string, node *v1.Node) (*v1.Node, error) { node = node.DeepCopy() if removalRequested, ok := node.Annotations[removalAnnotation]; ok { + // removal requires node name and address annotations; fail if either are not found + name, ok := node.Annotations[NodeNameAnnotation] + if !ok { + return node, fmt.Errorf("node name annotation for node %s not found", key) + } + address, ok := node.Annotations[NodeAddressAnnotation] + if !ok { + return node, fmt.Errorf("node address annotation for node %s not found", key) + } + lf := logrus.Fields{"name": name, "address": address} + + // Check to see if the node was previously removed from the cluster if removed, ok := node.Annotations[removedNodeNameAnnotation]; ok { - // check to see if removed is true. if it is, nothing to do. - if currentNodeName, ok := node.Annotations[NodeNameAnnotation]; ok { - if currentNodeName != removed { - // If the current node name is not the same as the removed node name, reset the tainted annotation and removed node name - logrus.Infof("Resetting removed node flag as removed node name (did not match current node name") - delete(node.Annotations, removedNodeNameAnnotation) - node.Annotations[removalAnnotation] = "false" - return e.nodeController.Update(node) - } - // this is the case where the current node name matches the removed node name. We have already removed the - // node, so no need to perform any action. Fallthrough to the non-op below. + if removed != name { + // If the current node name is not the same as the removed node name, clear the removal annotations, + // as this indicates that the node has been re-added with a new name. + logrus.WithFields(lf).Info("Resetting removed node flag as removed node name does not match current node name") + delete(node.Annotations, removedNodeNameAnnotation) + delete(node.Annotations, removalAnnotation) + return e.nodeController.Update(node) } - // This is the edge case where the removed annotation exists, but there is not a current node name annotation. - // This should be a non-op, as we can't remove the node anyway. - logrus.Debugf("etcd member %s was already marked via annotations as removed", key) + // Current node name matches removed node name; don't need to do anything return node, nil } - if strings.ToLower(removalRequested) == "true" { - // remove the member. - name, ok := node.Annotations[NodeNameAnnotation] - if !ok { - return node, fmt.Errorf("node name annotation for node %s not found", key) - } - address, ok := node.Annotations[NodeAddressAnnotation] - if !ok { - return node, fmt.Errorf("node address annotation for node %s not found", key) - } - logrus.Debugf("removing etcd member from cluster name: %s address: %s", name, address) + if strings.ToLower(removalRequested) == "true" { + // Removal requested, attempt to remove it from the cluster + logrus.WithFields(lf).Info("Removing etcd member from cluster due to remove annotation") if err := e.etcd.RemovePeer(e.ctx, name, address, true); err != nil { + // etcd will reject the removal if this is the only voting member; abort the removal by removing + // the annotation if this is the case. The requesting controller can re-request removal by setting + // the annotation again, once there are more cluster members. + if errors.Is(err, rpctypes.ErrMemberNotEnoughStarted) { + logrus.WithFields(lf).Errorf("etcd member removal rejected, clearing remove annotation: %v", err) + delete(node.Annotations, removalAnnotation) + return e.nodeController.Update(node) + } return node, err } - logrus.Debugf("etcd member removal successful for name: %s address: %s", name, address) - // Set the removed node name annotation and clean up the other etcd node annotations. - // These will be set if the tombstone file is then created and the etcd member is re-added, to their new - // respective values. + + logrus.WithFields(lf).Info("etcd emember removed successfully") + // Set the removed node name annotation and delete the etcd name and address annotations. + // These will be re-set to their new value when the member rejoins the cluster. node.Annotations[removedNodeNameAnnotation] = name delete(node.Annotations, NodeNameAnnotation) delete(node.Annotations, NodeAddressAnnotation) return e.nodeController.Update(node) } - // In the event that we had an unexpected removal value, simply return. + // In the event that we had an unexpected removal annotation value, simply return. // Fallthrough to the non-op below. } - // This is a non-op, as we don't have a tainted annotation to worry about. + // This is a non-op, as we don't have a deleted annotation to worry about. return node, nil } @@ -103,19 +111,13 @@ func (e *etcdMemberHandler) onRemove(key string, node *v1.Node) (*v1.Node, error logrus.Debugf("Node %s was not labeled etcd node, skipping etcd member removal", key) return node, nil } - logrus.Infof("Removing etcd member %s from cluster", key) - if removalRequested, ok := node.Annotations[removalAnnotation]; ok { - if strings.ToLower(removalRequested) == "true" { - if removedNodeName, ok := node.Annotations[removedNodeNameAnnotation]; ok { - if len(removedNodeName) > 0 { - // If we received a node to delete that has already been removed via annotation, it will be missing - // the corresponding node name and address annotations. - logrus.Infof("etcd member %s was already removed as member name %s via annotation from the cluster", key, removedNodeName) - return node, nil - } - } - } + + if removedNodeName, ok := node.Annotations[removedNodeNameAnnotation]; ok && len(removedNodeName) > 0 { + logrus.Debugf("Node %s was already removed from the cluster, skipping etcd member removal", key) + return node, nil } + + // removal requires node name and address annotations; fail if either are not found name, ok := node.Annotations[NodeNameAnnotation] if !ok { return node, fmt.Errorf("node name annotation for node %s not found", key) @@ -124,5 +126,8 @@ func (e *etcdMemberHandler) onRemove(key string, node *v1.Node) (*v1.Node, error if !ok { return node, fmt.Errorf("node address annotation for node %s not found", key) } + lf := logrus.Fields{"name": name, "address": address} + + logrus.WithFields(lf).Info("Removing etcd member from cluster due to node delete") return node, e.etcd.RemovePeer(e.ctx, name, address, true) } diff --git a/pkg/etcd/metadata_controller.go b/pkg/etcd/metadata_controller.go index fdf5ebed9a..da86e81263 100644 --- a/pkg/etcd/metadata_controller.go +++ b/pkg/etcd/metadata_controller.go @@ -3,12 +3,16 @@ package etcd import ( "context" "os" + "sync" "time" "github.com/k3s-io/k3s/pkg/util" controllerv1 "github.com/rancher/wrangler/pkg/generated/controllers/core/v1" "github.com/sirupsen/logrus" v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/client-go/util/retry" ) func registerMetadataHandlers(ctx context.Context, etcd *ETCD) { @@ -17,6 +21,7 @@ func registerMetadataHandlers(ctx context.Context, etcd *ETCD) { etcd: etcd, nodeController: nodes, ctx: ctx, + once: &sync.Once{}, } logrus.Infof("Starting managed etcd node metadata controller") @@ -27,9 +32,11 @@ type metadataHandler struct { etcd *ETCD nodeController controllerv1.NodeController ctx context.Context + once *sync.Once } func (m *metadataHandler) sync(key string, node *v1.Node) (*v1.Node, error) { + if node == nil { return nil, nil } @@ -48,6 +55,43 @@ func (m *metadataHandler) sync(key string, node *v1.Node) (*v1.Node, error) { return node, nil } +// checkReset ensures that member removal annotations are cleared when the cluster is reset. +// This is done here instead of in the member controller, as the member removal controller is +// not guaranteed to run on the node that was reset. +func (m *metadataHandler) checkReset() { + if resetDone, _ := m.etcd.IsReset(); resetDone { + labelSelector := labels.Set{util.ETCDRoleLabelKey: "true"}.String() + nodes, err := m.nodeController.List(metav1.ListOptions{LabelSelector: labelSelector}) + if err != nil { + logrus.Errorf("Failed to list etcd nodes: %v", err) + return + } + for _, n := range nodes.Items { + node := &n + err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + _, remove := node.Annotations[removalAnnotation] + _, removed := node.Annotations[removedNodeNameAnnotation] + if remove || removed { + node = node.DeepCopy() + delete(node.Annotations, removalAnnotation) + delete(node.Annotations, removedNodeNameAnnotation) + node, err = m.nodeController.Update(node) + return err + } + return nil + }) + if err != nil { + logrus.Errorf("Failed to clear removal annotations from node %s after cluster reset: %v", node.Name, err) + } else { + logrus.Infof("Cleared etcd member removal annotations from node %s after cluster reset", node.Name) + } + } + if err := m.etcd.clearReset(); err != nil { + logrus.Errorf("Failed to delete etcd cluster-reset file: %v", err) + } + } +} + func (m *metadataHandler) handleSelf(node *v1.Node) (*v1.Node, error) { if m.etcd.config.DisableETCD { if node.Annotations[NodeNameAnnotation] == "" && @@ -71,6 +115,8 @@ func (m *metadataHandler) handleSelf(node *v1.Node) (*v1.Node, error) { return m.nodeController.Update(node) } + m.once.Do(m.checkReset) + if node.Annotations[NodeNameAnnotation] == m.etcd.name && node.Annotations[NodeAddressAnnotation] == m.etcd.address && node.Labels[util.ETCDRoleLabelKey] == "true" { diff --git a/pkg/server/context.go b/pkg/server/context.go index 7af5122c90..49666e31be 100644 --- a/pkg/server/context.go +++ b/pkg/server/context.go @@ -38,21 +38,26 @@ func (c *Context) Start(ctx context.Context) error { return start.All(ctx, 5, c.K3s, c.Helm, c.Apps, c.Auth, c.Batch, c.Core) } -func NewContext(ctx context.Context, cfg string) (*Context, error) { +func NewContext(ctx context.Context, cfg string, forServer bool) (*Context, error) { restConfig, err := clientcmd.BuildConfigFromFlags("", cfg) if err != nil { return nil, err } restConfig.UserAgent = util.GetUserAgent(version.Program + "-supervisor") - if err := crds(ctx, restConfig); err != nil { - return nil, errors.Wrap(err, "failed to register CRDs") - } - k8s, err := kubernetes.NewForConfig(restConfig) if err != nil { return nil, err } + + var recorder record.EventRecorder + if forServer { + recorder = util.BuildControllerEventRecorder(k8s, version.Program+"-supervisor", metav1.NamespaceAll) + if err := crds(ctx, restConfig); err != nil { + return nil, errors.Wrap(err, "failed to register CRDs") + } + } + return &Context{ K3s: k3s.NewFactoryFromConfigOrDie(restConfig), Helm: helm.NewFactoryFromConfigOrDie(restConfig), @@ -61,7 +66,7 @@ func NewContext(ctx context.Context, cfg string) (*Context, error) { Apps: apps.NewFactoryFromConfigOrDie(restConfig), Batch: batch.NewFactoryFromConfigOrDie(restConfig), Core: core.NewFactoryFromConfigOrDie(restConfig), - Event: util.BuildControllerEventRecorder(k8s, version.Program+"-supervisor", metav1.NamespaceAll), + Event: recorder, }, nil } diff --git a/pkg/server/server.go b/pkg/server/server.go index 834041595f..e18001a51a 100644 --- a/pkg/server/server.go +++ b/pkg/server/server.go @@ -98,7 +98,7 @@ func startOnAPIServerReady(ctx context.Context, config *Config) { func runControllers(ctx context.Context, config *Config) error { controlConfig := &config.ControlConfig - sc, err := NewContext(ctx, controlConfig.Runtime.KubeConfigSupervisor) + sc, err := NewContext(ctx, controlConfig.Runtime.KubeConfigSupervisor, true) if err != nil { return errors.Wrap(err, "failed to create new server context") }