Skip to content

K
k3s
Commit 002e6c43 authored by Brad Davidson's avatar Brad Davidson Committed by Brad Davidson
Browse files
Options

Reorganize Driver interface and etcd driver to avoid passing context and config into most calls

Signed-off-by: 's avatarBrad Davidson <brad.davidson@rancher.com>
parent 89064592
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 200 additions and 157 deletions
pkg/cli/etcdsnapshot/etcd_snapshot.go
View file @ 002e6c43
......@@ -92,11 +92,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
}
......@@ -140,7 +140,7 @@ 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
}
......@@ -180,7 +180,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
}
......@@ -246,7 +246,7 @@ 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
}
......
pkg/cluster/bootstrap.go
View file @ 002e6c43
......@@ -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
View file @ 002e6c43
......@@ -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
View file @ 002e6c43
......@@ -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
View file @ 002e6c43
......@@ -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_test.go
View file @ 002e6c43
......@@ -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
......@@ -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)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment