[Release-1.25] Clear remove annotations on cluster reset (#8589)

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
 }
@@ -97,11 +97,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
 	}
@@ -115,7 +115,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
 	}
@@ -145,11 +145,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
 	}
@@ -185,7 +185,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
 	}
 
@@ -251,11 +251,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
 	}

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 {

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
 	}

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,75 +55,64 @@ 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
-		} else if ok {
-			c.managedDB = driver
-			return nil
 		}
-	}
-
-	// 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() {
+		if ok, err := driver.IsInitialized(); err != nil {
+			return err
+		} else if ok {
 			c.managedDB = driver
 			return nil
 		}
@@ -133,12 +120,7 @@ func (c *Cluster) assignManagedDriver(ctx context.Context) error {
 
 	// 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

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]
 }

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 {
-	e.config = config
-
-	client, err := GetClient(ctx, e.config)
-	if err != nil {
-		return err
+// 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.client = client
 
-	go func() {
-		<-ctx.Done()
-		e.client.Close()
-	}()
+	e.config = config
 
-	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(),
@@ -2111,7 +2175,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
 	}
@@ -2165,8 +2229,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)
 }

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)

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)
 }

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" {

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
 }
 

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")
 	}