Skip to content

K
k3s
Commit 61bbad7d authored by Brad Davidson's avatar Brad Davidson Committed by Brad Davidson
Browse files
Options

Store extra metadata and cluster ID for snapshots

Write the extra metadata both locally and to S3. These files are placed such that they will not be used by older versions of K3s that do not make use of them. Signed-off-by: 's avatarBrad Davidson <brad.davidson@rancher.com> (cherry picked from commit 74640070) Signed-off-by: 's avatarBrad Davidson <brad.davidson@rancher.com>
parent db4ee1b2
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 155 additions and 197 deletions
pkg/cli/etcdsnapshot/etcd_snapshot.go
View file @ 61bbad7d
package etcdsnapshot
import (
"context"
"encoding/json"
"errors"
"fmt"
......@@ -12,8 +13,7 @@ import (
"github.com/erikdubbelboer/gspt"
"github.com/k3s-io/k3s/pkg/cli/cmds"
"github.com/k3s-io/k3s/pkg/cluster"
"github.com/k3s-io/k3s/pkg/daemons/config"
daemonconfig "github.com/k3s-io/k3s/pkg/daemons/config"
"github.com/k3s-io/k3s/pkg/etcd"
"github.com/k3s-io/k3s/pkg/server"
util2 "github.com/k3s-io/k3s/pkg/util"
......@@ -22,16 +22,22 @@ import (
"gopkg.in/yaml.v2"
)
type etcdCommand struct {
etcd *etcd.ETCD
ctx context.Context
}
// commandSetup setups up common things needed
// for each etcd command.
func commandSetup(app *cli.Context, cfg *cmds.Server, sc *server.Config) error {
func commandSetup(app *cli.Context, cfg *cmds.Server, config *server.Config) (*etcdCommand, error) {
ctx := signals.SetupSignalContext()
gspt.SetProcTitle(os.Args[0])
nodeName := app.String("node-name")
if nodeName == "" {
h, err := os.Hostname()
if err != nil {
return err
return nil, err
}
nodeName = h
}
......@@ -40,33 +46,53 @@ func commandSetup(app *cli.Context, cfg *cmds.Server, sc *server.Config) error {
dataDir, err := server.ResolveDataDir(cfg.DataDir)
if err != nil {
return err
return nil, err
}
config.DisableAgent = true
config.ControlConfig.DataDir = dataDir
config.ControlConfig.EtcdSnapshotName = cfg.EtcdSnapshotName
config.ControlConfig.EtcdSnapshotDir = cfg.EtcdSnapshotDir
config.ControlConfig.EtcdSnapshotCompress = cfg.EtcdSnapshotCompress
config.ControlConfig.EtcdListFormat = strings.ToLower(cfg.EtcdListFormat)
config.ControlConfig.EtcdS3 = cfg.EtcdS3
config.ControlConfig.EtcdS3Endpoint = cfg.EtcdS3Endpoint
config.ControlConfig.EtcdS3EndpointCA = cfg.EtcdS3EndpointCA
config.ControlConfig.EtcdS3SkipSSLVerify = cfg.EtcdS3SkipSSLVerify
config.ControlConfig.EtcdS3AccessKey = cfg.EtcdS3AccessKey
config.ControlConfig.EtcdS3SecretKey = cfg.EtcdS3SecretKey
config.ControlConfig.EtcdS3BucketName = cfg.EtcdS3BucketName
config.ControlConfig.EtcdS3Region = cfg.EtcdS3Region
config.ControlConfig.EtcdS3Folder = cfg.EtcdS3Folder
config.ControlConfig.EtcdS3Insecure = cfg.EtcdS3Insecure
config.ControlConfig.EtcdS3Timeout = cfg.EtcdS3Timeout
config.ControlConfig.Runtime = daemonconfig.NewRuntime(nil)
config.ControlConfig.Runtime.ETCDServerCA = filepath.Join(dataDir, "tls", "etcd", "server-ca.crt")
config.ControlConfig.Runtime.ClientETCDCert = filepath.Join(dataDir, "tls", "etcd", "client.crt")
config.ControlConfig.Runtime.ClientETCDKey = filepath.Join(dataDir, "tls", "etcd", "client.key")
config.ControlConfig.Runtime.KubeConfigAdmin = filepath.Join(dataDir, "cred", "admin.kubeconfig")
e := etcd.NewETCD()
if err := e.SetControlConfig(&config.ControlConfig); err != nil {
return nil, err
}
initialized, err := e.IsInitialized()
if err != nil {
return nil, err
}
if !initialized {
return nil, fmt.Errorf("etcd database not found in %s", config.ControlConfig.DataDir)
}
sc.DisableAgent = true
sc.ControlConfig.DataDir = dataDir
sc.ControlConfig.EtcdSnapshotName = cfg.EtcdSnapshotName
sc.ControlConfig.EtcdSnapshotDir = cfg.EtcdSnapshotDir
sc.ControlConfig.EtcdSnapshotCompress = cfg.EtcdSnapshotCompress
sc.ControlConfig.EtcdListFormat = strings.ToLower(cfg.EtcdListFormat)
sc.ControlConfig.EtcdS3 = cfg.EtcdS3
sc.ControlConfig.EtcdS3Endpoint = cfg.EtcdS3Endpoint
sc.ControlConfig.EtcdS3EndpointCA = cfg.EtcdS3EndpointCA
sc.ControlConfig.EtcdS3SkipSSLVerify = cfg.EtcdS3SkipSSLVerify
sc.ControlConfig.EtcdS3AccessKey = cfg.EtcdS3AccessKey
sc.ControlConfig.EtcdS3SecretKey = cfg.EtcdS3SecretKey
sc.ControlConfig.EtcdS3BucketName = cfg.EtcdS3BucketName
sc.ControlConfig.EtcdS3Region = cfg.EtcdS3Region
sc.ControlConfig.EtcdS3Folder = cfg.EtcdS3Folder
sc.ControlConfig.EtcdS3Insecure = cfg.EtcdS3Insecure
sc.ControlConfig.EtcdS3Timeout = cfg.EtcdS3Timeout
sc.ControlConfig.Runtime = config.NewRuntime(nil)
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.KubeConfigAdmin = filepath.Join(dataDir, "cred", "admin.kubeconfig")
sc, err := server.NewContext(ctx, config.ControlConfig.Runtime.KubeConfigAdmin, false)
if err != nil {
return nil, err
}
config.ControlConfig.Runtime.K3s = sc.K3s
config.ControlConfig.Runtime.Core = sc.Core
return nil
return &etcdCommand{etcd: e, ctx: ctx}, nil
}
// Run is an alias for Save, retained for compatibility reasons.
......@@ -85,43 +111,18 @@ func Save(app *cli.Context) error {
func save(app *cli.Context, cfg *cmds.Server) error {
var serverConfig server.Config
if err := commandSetup(app, cfg, &serverConfig); err != nil {
return err
}
if len(app.Args()) > 0 {
return util2.ErrCommandNoArgs
}
serverConfig.ControlConfig.EtcdSnapshotRetention = 0 // disable retention check
ctx := signals.SetupSignalContext()
e := etcd.NewETCD()
if err := e.SetControlConfig(&serverConfig.ControlConfig); err != nil {
return err
}
initialized, err := e.IsInitialized()
ec, err := commandSetup(app, cfg, &serverConfig)
if err != nil {
return err
}
if !initialized {
return fmt.Errorf("etcd database not found in %s", serverConfig.ControlConfig.DataDir)
}
cluster := cluster.New(&serverConfig.ControlConfig)
if err := cluster.Bootstrap(ctx, true); err != nil {
return err
}
sc, err := server.NewContext(ctx, serverConfig.ControlConfig.Runtime.KubeConfigAdmin, false)
if err != nil {
return err
}
serverConfig.ControlConfig.Runtime.Core = sc.Core
serverConfig.ControlConfig.EtcdSnapshotRetention = 0 // disable retention check
return cluster.Snapshot(ctx, &serverConfig.ControlConfig)
return ec.etcd.Snapshot(ec.ctx)
}
func Delete(app *cli.Context) error {
......@@ -134,7 +135,8 @@ func Delete(app *cli.Context) error {
func delete(app *cli.Context, cfg *cmds.Server) error {
var serverConfig server.Config
if err := commandSetup(app, cfg, &serverConfig); err != nil {
ec, err := commandSetup(app, cfg, &serverConfig)
if err != nil {
return err
}
......@@ -143,19 +145,7 @@ func delete(app *cli.Context, cfg *cmds.Server) error {
return errors.New("no snapshots given for removal")
}
ctx := signals.SetupSignalContext()
e := etcd.NewETCD()
if err := e.SetControlConfig(&serverConfig.ControlConfig); err != nil {
return err
}
sc, err := server.NewContext(ctx, serverConfig.ControlConfig.Runtime.KubeConfigAdmin, false)
if err != nil {
return err
}
serverConfig.ControlConfig.Runtime.Core = sc.Core
return e.DeleteSnapshots(ctx, app.Args())
return ec.etcd.DeleteSnapshots(ec.ctx, app.Args())
}
func List(app *cli.Context) error {
......@@ -165,7 +155,7 @@ func List(app *cli.Context) error {
return list(app, &cmds.ServerConfig)
}
var etcdListFormats = []string{"json", "yaml"}
var etcdListFormats = []string{"json", "yaml", "table"}
func validEtcdListFormat(format string) bool {
for _, supportedFormat := range etcdListFormats {
......@@ -179,17 +169,12 @@ func validEtcdListFormat(format string) bool {
func list(app *cli.Context, cfg *cmds.Server) error {
var serverConfig server.Config
if err := commandSetup(app, cfg, &serverConfig); err != nil {
return err
}
ctx := signals.SetupSignalContext()
e := etcd.NewETCD()
if err := e.SetControlConfig(&serverConfig.ControlConfig); err != nil {
ec, err := commandSetup(app, cfg, &serverConfig)
if err != nil {
return err
}
sf, err := e.ListSnapshots(ctx)
sf, err := ec.etcd.ListSnapshots(ec.ctx)
if err != nil {
return err
}
......@@ -213,20 +198,9 @@ func list(app *cli.Context, cfg *cmds.Server) error {
w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0)
defer w.Flush()
if cfg.EtcdS3 {
fmt.Fprint(w, "Name\tSize\tCreated\n")
for _, s := range sf {
if s.NodeName == "s3" {
fmt.Fprintf(w, "%s\t%d\t%s\n", s.Name, s.Size, s.CreatedAt.Format(time.RFC3339))
}
}
} else {
fmt.Fprint(w, "Name\tLocation\tSize\tCreated\n")
for _, s := range sf {
if s.NodeName != "s3" {
fmt.Fprintf(w, "%s\t%s\t%d\t%s\n", s.Name, s.Location, s.Size, s.CreatedAt.Format(time.RFC3339))
}
}
fmt.Fprint(w, "Name\tLocation\tSize\tCreated\n")
for _, s := range sf {
fmt.Fprintf(w, "%s\t%s\t%d\t%s\n", s.Name, s.Location, s.Size, s.CreatedAt.Format(time.RFC3339))
}
}
......@@ -243,23 +217,12 @@ func Prune(app *cli.Context) error {
func prune(app *cli.Context, cfg *cmds.Server) error {
var serverConfig server.Config
if err := commandSetup(app, cfg, &serverConfig); err != nil {
ec, err := commandSetup(app, cfg, &serverConfig)
if err != nil {
return err
}
serverConfig.ControlConfig.EtcdSnapshotRetention = cfg.EtcdSnapshotRetention
ctx := signals.SetupSignalContext()
e := etcd.NewETCD()
if err := e.SetControlConfig(&serverConfig.ControlConfig); err != nil {
return err
}
sc, err := server.NewContext(ctx, serverConfig.ControlConfig.Runtime.KubeConfigAdmin, false)
if err != nil {
return err
}
serverConfig.ControlConfig.Runtime.Core = sc.Core
return e.PruneSnapshots(ctx)
return ec.etcd.PruneSnapshots(ec.ctx)
}
pkg/cluster/bootstrap.go
View file @ 61bbad7d
......@@ -424,15 +424,6 @@ func (c *Cluster) bootstrap(ctx context.Context) error {
return c.storageBootstrap(ctx)
}
// Snapshot is a proxy method to call the snapshot method on the managedb
// interface for etcd clusters.
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)
}
// compareConfig verifies that the config of the joining control plane node coincides with the cluster's config
func (c *Cluster) compareConfig() error {
token := c.config.AgentToken
......
pkg/cluster/bootstrap_test.go
View file @ 61bbad7d
......@@ -197,50 +197,3 @@ func TestCluster_migrateBootstrapData(t *testing.T) {
})
}
}
func TestCluster_Snapshot(t *testing.T) {
type fields struct {
clientAccessInfo *clientaccess.Info
config *config.Control
managedDB managed.Driver
joining bool
storageStarted bool
saveBootstrap bool
shouldBootstrap bool
}
type args struct {
ctx context.Context
config *config.Control
}
tests := []struct {
name string
fields fields
args args
wantErr bool
}{
{
name: "Fail on non etcd cluster",
fields: fields{},
args: args{
ctx: context.Background(),
},
wantErr: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
c := &Cluster{
clientAccessInfo: tt.fields.clientAccessInfo,
config: tt.fields.config,
managedDB: tt.fields.managedDB,
joining: tt.fields.joining,
storageStarted: tt.fields.storageStarted,
saveBootstrap: tt.fields.saveBootstrap,
shouldBootstrap: tt.fields.shouldBootstrap,
}
if err := c.Snapshot(tt.args.ctx, tt.args.config); (err != nil) != tt.wantErr {
t.Errorf("Cluster.Snapshot() error = %v, wantErr %v", err, tt.wantErr)
}
})
}
}
pkg/daemons/config/types.go
View file @ 61bbad7d
......@@ -10,6 +10,7 @@ import (
"sync"
"time"
"github.com/k3s-io/k3s/pkg/generated/controllers/k3s.cattle.io"
"github.com/k3s-io/kine/pkg/endpoint"
"github.com/rancher/wrangler/pkg/generated/controllers/core"
"github.com/rancher/wrangler/pkg/leader"
......@@ -341,6 +342,7 @@ type ControlRuntime struct {
ClientETCDCert string
ClientETCDKey string
K3s *k3s.Factory
Core *core.Factory
Event record.EventRecorder
EtcdConfig endpoint.ETCDConfig
......
pkg/etcd/snapshot.go
View file @ 61bbad7d
......@@ -38,6 +38,7 @@ const (
maxConcurrentSnapshots = 1
pruneStepSize = 5
compressedExtension = ".zip"
metadataDir = ".metadata"
)
var (
......@@ -272,20 +273,20 @@ func (e *ETCD) Snapshot(ctx context.Context) error {
}
}
if e.config.EtcdSnapshotCompress {
zipPath, err := e.compressSnapshot(snapshotDir, snapshotName, snapshotPath, now)
if err != nil {
return err
}
if err := os.Remove(snapshotPath); err != nil {
return err
// If the snapshot attempt was successful, sf will be nil as we did not set it to store the error message.
if sf == nil {
if e.config.EtcdSnapshotCompress {
zipPath, err := e.compressSnapshot(snapshotDir, snapshotName, snapshotPath, now)
if err != nil {
return errors.Wrap(err, "failed to compress snapshot")
}
if err := os.Remove(snapshotPath); err != nil {
return errors.Wrap(err, "failed to remove uncompressed snapshot")
}
snapshotPath = zipPath
logrus.Info("Compressed snapshot: " + snapshotPath)
}
snapshotPath = zipPath
logrus.Info("Compressed snapshot: " + snapshotPath)
}
// If the snapshot attempt was successful, sf will be nil as we did not set it.
if sf == nil {
f, err := os.Stat(snapshotPath)
if err != nil {
return errors.Wrap(err, "unable to retrieve snapshot information from local snapshot")
......@@ -303,15 +304,19 @@ func (e *ETCD) Snapshot(ctx context.Context) error {
metadataSource: extraMetadata,
}
if err := saveSnapshotMetadata(snapshotPath, extraMetadata); err != nil {
return errors.Wrap(err, "failed to save local snapshot metadata")
}
if err := e.addSnapshotData(*sf); err != nil {
return errors.Wrap(err, "failed to save local snapshot data to configmap")
}
if err := snapshotRetention(e.config.EtcdSnapshotRetention, e.config.EtcdSnapshotName, snapshotDir); err != nil {
return errors.Wrap(err, "failed to apply local snapshot retention policy")
}
if e.config.EtcdS3 {
logrus.Infof("Saving etcd snapshot %s to S3", snapshotName)
if err := e.initS3IfNil(ctx); err != nil {
logrus.Warnf("Unable to initialize S3 client: %v", err)
sf = &snapshotFile{
......@@ -335,6 +340,7 @@ func (e *ETCD) Snapshot(ctx context.Context) error {
metadataSource: extraMetadata,
}
} else {
logrus.Infof("Saving etcd snapshot %s to S3", snapshotName)
// upload will return a snapshotFile even on error - if there was an
// error, it will be reflected in the status and message.
sf, err = e.s3.upload(ctx, snapshotPath, extraMetadata, now)
......@@ -414,10 +420,21 @@ func (e *ETCD) listLocalSnapshots() (map[string]snapshotFile, error) {
if err != nil {
ts = file.ModTime().Unix()
}
// try to read metadata from disk; don't warn if it is missing as it will not exist
// for snapshot files from old releases or if there was no metadata provided.
var metadata string
metadataFile := filepath.Join(filepath.Dir(path), "..", metadataDir, file.Name())
if m, err := os.ReadFile(metadataFile); err == nil {
logrus.Debugf("Loading snapshot metadata from %s", metadataFile)
metadata = base64.StdEncoding.EncodeToString(m)
}
sf := snapshotFile{
Name: file.Name(),
Location: "file://" + filepath.Join(snapshotDir, file.Name()),
NodeName: nodeName,
Metadata: metadata,
CreatedAt: &metav1.Time{
Time: time.Unix(ts, 0),
},
......@@ -462,7 +479,7 @@ func (e *ETCD) PruneSnapshots(ctx context.Context) error {
if e.config.EtcdS3 {
if err := e.initS3IfNil(ctx); err != nil {
logrus.Warnf("Unable to initialize S3 client during prune: %v", err)
logrus.Warnf("Unable to initialize S3 client: %v", err)
} else {
if err := e.s3.snapshotRetention(ctx); err != nil {
logrus.Errorf("Error applying S3 snapshot retention policy: %v", err)
......@@ -478,6 +495,7 @@ func (e *ETCD) ListSnapshots(ctx context.Context) (map[string]snapshotFile, erro
snapshotFiles := map[string]snapshotFile{}
if e.config.EtcdS3 {
if err := e.initS3IfNil(ctx); err != nil {
logrus.Warnf("Unable to initialize S3 client: %v", err)
return nil, err
}
sfs, err := e.s3.listSnapshots(ctx)
......@@ -506,13 +524,30 @@ func (e *ETCD) DeleteSnapshots(ctx context.Context, snapshots []string) error {
return errors.Wrap(err, "failed to get the snapshot dir")
}
if e.config.EtcdS3 {
logrus.Info("Removing the given etcd snapshot(s) from S3")
logrus.Debugf("Removing the given etcd snapshot(s) from S3: %v", snapshots)
logrus.Info("Removing the given locally stored etcd snapshot(s)")
logrus.Debugf("Attempting to remove the given locally stored etcd snapshot(s): %v", snapshots)
for _, s := range snapshots {
// check if the given snapshot exists. If it does,
// remove it, otherwise continue.
sf := filepath.Join(snapshotDir, s)
if _, err := os.Stat(sf); os.IsNotExist(err) {
logrus.Infof("Snapshot %s, does not exist", s)
continue
}
if err := os.Remove(sf); err != nil {
return err
}
logrus.Debug("Removed snapshot ", s)
}
if e.config.EtcdS3 {
if e.initS3IfNil(ctx); err != nil {
logrus.Warnf("Unable to initialize S3 client: %v", err)
return err
}
logrus.Info("Removing the given etcd snapshot(s) from S3")
logrus.Debugf("Removing the given etcd snapshot(s) from S3: %v", snapshots)
objectsCh := make(chan minio.ObjectInfo)
......@@ -566,23 +601,6 @@ func (e *ETCD) DeleteSnapshots(ctx context.Context, snapshots []string) error {
}
}
logrus.Info("Removing the given locally stored etcd snapshot(s)")
logrus.Debugf("Attempting to remove the given locally stored etcd snapshot(s): %v", snapshots)
for _, s := range snapshots {
// check if the given snapshot exists. If it does,
// remove it, otherwise continue.
sf := filepath.Join(snapshotDir, s)
if _, err := os.Stat(sf); os.IsNotExist(err) {
logrus.Infof("Snapshot %s, does not exist", s)
continue
}
if err := os.Remove(sf); err != nil {
return err
}
logrus.Debug("Removed snapshot ", s)
}
return e.ReconcileSnapshotData(ctx)
}
......@@ -735,6 +753,11 @@ func (e *ETCD) ReconcileSnapshotData(ctx context.Context) error {
var s3ListSuccessful bool
if e.config.EtcdS3 {
if err := e.initS3IfNil(ctx); err != nil {
logrus.Warnf("Unable to initialize S3 client: %v", err)
return err
}
if s3Snapshots, err := e.s3.listSnapshots(ctx); err != nil {
logrus.Errorf("Error retrieving S3 snapshots for reconciliation: %v", err)
} else {
......@@ -906,10 +929,14 @@ func snapshotRetention(retention int, snapshotPrefix string, snapshotDir string)
for _, df := range snapshotFiles[retention:] {
snapshotPath := filepath.Join(snapshotDir, df.Name)
metadataPath := filepath.Join(snapshotDir, "..", metadataDir, df.Name)
logrus.Infof("Removing local snapshot %s", snapshotPath)
if err := os.Remove(snapshotPath); err != nil {
return err
}
if err := os.Remove(metadataPath); err != nil && !os.IsNotExist(err) {
return err
}
}
return nil
......@@ -919,3 +946,24 @@ func isTooLargeError(err error) bool {
// There are no helpers for unpacking field validation errors, so we just check for "Too long" in the error string.
return apierrors.IsRequestEntityTooLargeError(err) || (apierrors.IsInvalid(err) && strings.Contains(err.Error(), "Too long"))
}
// saveSnapshotMetadata writes extra metadata to disk.
// The upload is silently skipped if no extra metadata is provided.
func saveSnapshotMetadata(snapshotPath string, extraMetadata *v1.ConfigMap) error {
if extraMetadata == nil || len(extraMetadata.Data) == 0 {
return nil
}
dir := filepath.Join(filepath.Dir(snapshotPath), "..", metadataDir)
filename := filepath.Base(snapshotPath)
metadataPath := filepath.Join(dir, filename)
logrus.Infof("Saving snapshot metadata to %s", metadataPath)
m, err := json.Marshal(extraMetadata.Data)
if err != nil {
return err
}
if err := os.MkdirAll(dir, 0700); err != nil {
return err
}
return os.WriteFile(metadataPath, m, 0700)
}
pkg/server/server.go
View file @ 61bbad7d
......@@ -115,6 +115,7 @@ func runControllers(ctx context.Context, config *Config) error {
controlConfig.Runtime.NodePasswdFile); err != nil {
logrus.Warn(errors.Wrap(err, "error migrating node-password file"))
}
controlConfig.Runtime.K3s = sc.K3s
controlConfig.Runtime.Event = sc.Event
controlConfig.Runtime.Core = sc.Core
......
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