Commit bcaa6392 authored by Brad Davidson's avatar Brad Davidson Committed by Brad Davidson

Add etcd s3 config secret implementation

* Move snapshot structs and functions into pkg/etcd/snapshot * Move s3 client code and functions into pkg/etcd/s3 * Refactor pkg/etcd to track snapshot and s3 moves * Add support for reading s3 client config from secret * Add minio client cache, since S3 client configuration can now be changed at runtime by modifying the secret, and don't want to have to create a new minio client every time we read config. * Add tests for pkg/etcd/s3 Signed-off-by: 's avatarBrad Davidson <brad.davidson@rancher.com> (cherry picked from commit c36db53e) Signed-off-by: 's avatarBrad Davidson <brad.davidson@rancher.com>
parent c338b313
# Support etcd Snapshot Configuration via Kubernetes Secret
Date: 2024-02-06
Revised: 2024-06-10
## Status
......@@ -34,8 +35,8 @@ avoids embedding the credentials directly in the system configuration, chart val
* We will add a `--etcd-s3-proxy` flag that can be used to set the proxy used by the S3 client. This will override the
settings that golang's default HTTP client reads from the `HTTP_PROXY/HTTPS_PROXY/NO_PROXY` environment varibles.
* We will add support for reading etcd snapshot S3 configuration from a Secret. The secret name will be specified via a new
`--etcd-s3-secret` flag, which accepts the name of the Secret in the `kube-system` namespace.
* Presence of the `--etcd-s3-secret` flag does not imply `--etcd-s3`. If S3 is not enabled by use of the `--etcd-s3` flag,
`--etcd-s3-config-secret` flag, which accepts the name of the Secret in the `kube-system` namespace.
* Presence of the `--etcd-s3-config-secret` flag does not imply `--etcd-s3`. If S3 is not enabled by use of the `--etcd-s3` flag,
the Secret will not be used.
* The Secret does not need to exist when K3s starts; it will be checked for every time a snapshot operation is performed.
* Secret and CLI/config values will NOT be merged. The Secret will provide values to be used in absence of other
......@@ -64,6 +65,7 @@ stringData:
etcd-s3-access-key: "AWS_ACCESS_KEY_ID"
etcd-s3-secret-key: "AWS_SECRET_ACCESS_KEY"
etcd-s3-bucket: "bucket"
etcd-s3-folder: "folder"
etcd-s3-region: "us-east-1"
etcd-s3-insecure: "false"
etcd-s3-timeout: "5m"
......@@ -73,3 +75,9 @@ stringData:
## Consequences
This will require additional documentation, tests, and QA work to validate use of secrets for s3 snapshot configuration.
## Revisions
#### 2024-06-10:
* Changed flag to `etcd-s3-config-secret` to avoid confusion with `etcd-s3-secret-key`.
* Added `etcd-s3-folder` to example Secret.
......@@ -137,11 +137,9 @@ require (
github.com/vishvananda/netlink v1.2.1-beta.2
github.com/yl2chen/cidranger v1.0.2
go.etcd.io/etcd/api/v3 v3.5.13
go.etcd.io/etcd/client/pkg/v3 v3.5.13
go.etcd.io/etcd/client/v3 v3.5.13
go.etcd.io/etcd/etcdutl/v3 v3.5.9
go.etcd.io/etcd/etcdutl/v3 v3.5.13
go.etcd.io/etcd/server/v3 v3.5.13
go.uber.org/zap v1.27.0
golang.org/x/crypto v0.23.0
golang.org/x/net v0.25.0
golang.org/x/sync v0.7.0
......@@ -416,6 +414,7 @@ require (
github.com/xlab/treeprint v1.2.0 // indirect
github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
go.etcd.io/bbolt v1.3.9 // indirect
go.etcd.io/etcd/client/pkg/v3 v3.5.13 // indirect
go.etcd.io/etcd/client/v2 v2.305.13 // indirect
go.etcd.io/etcd/pkg/v3 v3.5.13 // indirect
go.etcd.io/etcd/raft/v3 v3.5.13 // indirect
......@@ -436,6 +435,7 @@ require (
go.uber.org/fx v1.20.1 // indirect
go.uber.org/mock v0.4.0 // indirect
go.uber.org/multierr v1.11.0 // indirect
go.uber.org/zap v1.27.0 // indirect
golang.org/x/exp v0.0.0-20240222234643-814bf88cf225 // indirect
golang.org/x/mod v0.17.0 // indirect
golang.org/x/oauth2 v0.17.0 // indirect
......
......@@ -100,6 +100,16 @@ var EtcdSnapshotFlags = []cli.Flag{
Usage: "(db) S3 folder",
Destination: &ServerConfig.EtcdS3Folder,
},
&cli.StringFlag{
Name: "s3-proxy,etcd-s3-proxy",
Usage: "(db) Proxy server to use when connecting to S3, overriding any proxy-releated environment variables",
Destination: &ServerConfig.EtcdS3Proxy,
},
&cli.StringFlag{
Name: "s3-config-secret,etcd-s3-config-secret",
Usage: "(db) Name of secret in the kube-system namespace used to configure S3, if etcd-s3 is enabled and no other etcd-s3 options are set",
Destination: &ServerConfig.EtcdS3ConfigSecret,
},
&cli.BoolFlag{
Name: "s3-insecure,etcd-s3-insecure",
Usage: "(db) Disables S3 over HTTPS",
......
......@@ -104,6 +104,8 @@ type Server struct {
EtcdS3BucketName string
EtcdS3Region string
EtcdS3Folder string
EtcdS3Proxy string
EtcdS3ConfigSecret string
EtcdS3Timeout time.Duration
EtcdS3Insecure bool
ServiceLBNamespace string
......@@ -430,6 +432,16 @@ var ServerFlags = []cli.Flag{
Usage: "(db) S3 folder",
Destination: &ServerConfig.EtcdS3Folder,
},
&cli.StringFlag{
Name: "etcd-s3-proxy",
Usage: "(db) Proxy server to use when connecting to S3, overriding any proxy-releated environment variables",
Destination: &ServerConfig.EtcdS3Proxy,
},
&cli.StringFlag{
Name: "etcd-s3-config-secret",
Usage: "(db) Name of secret in the kube-system namespace used to configure S3, if etcd-s3 is enabled and no other etcd-s3 options are set",
Destination: &ServerConfig.EtcdS3ConfigSecret,
},
&cli.BoolFlag{
Name: "etcd-s3-insecure",
Usage: "(db) Disables S3 over HTTPS",
......
......@@ -16,6 +16,7 @@ import (
"github.com/k3s-io/k3s/pkg/cli/cmds"
"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/etcd"
"github.com/k3s-io/k3s/pkg/proctitle"
"github.com/k3s-io/k3s/pkg/server"
......@@ -50,17 +51,20 @@ func commandSetup(app *cli.Context, cfg *cmds.Server) (*etcd.SnapshotRequest, *c
}
if cfg.EtcdS3 {
sr.S3 = &etcd.SnapshotRequestS3{}
sr.S3.AccessKey = cfg.EtcdS3AccessKey
sr.S3.Bucket = cfg.EtcdS3BucketName
sr.S3.Endpoint = cfg.EtcdS3Endpoint
sr.S3.EndpointCA = cfg.EtcdS3EndpointCA
sr.S3.Folder = cfg.EtcdS3Folder
sr.S3.Insecure = cfg.EtcdS3Insecure
sr.S3.Region = cfg.EtcdS3Region
sr.S3.SecretKey = cfg.EtcdS3SecretKey
sr.S3.SkipSSLVerify = cfg.EtcdS3SkipSSLVerify
sr.S3.Timeout = metav1.Duration{Duration: cfg.EtcdS3Timeout}
sr.S3 = &config.EtcdS3{
AccessKey: cfg.EtcdS3AccessKey,
Bucket: cfg.EtcdS3BucketName,
ConfigSecret: cfg.EtcdS3ConfigSecret,
Endpoint: cfg.EtcdS3Endpoint,
EndpointCA: cfg.EtcdS3EndpointCA,
Folder: cfg.EtcdS3Folder,
Insecure: cfg.EtcdS3Insecure,
Proxy: cfg.EtcdS3Proxy,
Region: cfg.EtcdS3Region,
SecretKey: cfg.EtcdS3SecretKey,
SkipSSLVerify: cfg.EtcdS3SkipSSLVerify,
Timeout: metav1.Duration{Duration: cfg.EtcdS3Timeout},
}
// extend request timeout to allow the S3 operation to complete
timeout += cfg.EtcdS3Timeout
}
......
......@@ -32,6 +32,7 @@ import (
"github.com/rancher/wrangler/v3/pkg/signals"
"github.com/sirupsen/logrus"
"github.com/urfave/cli"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
utilnet "k8s.io/apimachinery/pkg/util/net"
kubeapiserverflag "k8s.io/component-base/cli/flag"
"k8s.io/kubernetes/pkg/controlplane/apiserver/options"
......@@ -186,17 +187,22 @@ func run(app *cli.Context, cfg *cmds.Server, leaderControllers server.CustomCont
serverConfig.ControlConfig.EtcdSnapshotCron = cfg.EtcdSnapshotCron
serverConfig.ControlConfig.EtcdSnapshotDir = cfg.EtcdSnapshotDir
serverConfig.ControlConfig.EtcdSnapshotRetention = cfg.EtcdSnapshotRetention
serverConfig.ControlConfig.EtcdS3 = cfg.EtcdS3
serverConfig.ControlConfig.EtcdS3Endpoint = cfg.EtcdS3Endpoint
serverConfig.ControlConfig.EtcdS3EndpointCA = cfg.EtcdS3EndpointCA
serverConfig.ControlConfig.EtcdS3SkipSSLVerify = cfg.EtcdS3SkipSSLVerify
serverConfig.ControlConfig.EtcdS3AccessKey = cfg.EtcdS3AccessKey
serverConfig.ControlConfig.EtcdS3SecretKey = cfg.EtcdS3SecretKey
serverConfig.ControlConfig.EtcdS3BucketName = cfg.EtcdS3BucketName
serverConfig.ControlConfig.EtcdS3Region = cfg.EtcdS3Region
serverConfig.ControlConfig.EtcdS3Folder = cfg.EtcdS3Folder
serverConfig.ControlConfig.EtcdS3Insecure = cfg.EtcdS3Insecure
serverConfig.ControlConfig.EtcdS3Timeout = cfg.EtcdS3Timeout
if cfg.EtcdS3 {
serverConfig.ControlConfig.EtcdS3 = &config.EtcdS3{
AccessKey: cfg.EtcdS3AccessKey,
Bucket: cfg.EtcdS3BucketName,
ConfigSecret: cfg.EtcdS3ConfigSecret,
Endpoint: cfg.EtcdS3Endpoint,
EndpointCA: cfg.EtcdS3EndpointCA,
Folder: cfg.EtcdS3Folder,
Insecure: cfg.EtcdS3Insecure,
Proxy: cfg.EtcdS3Proxy,
Region: cfg.EtcdS3Region,
SecretKey: cfg.EtcdS3SecretKey,
SkipSSLVerify: cfg.EtcdS3SkipSSLVerify,
Timeout: metav1.Duration{Duration: cfg.EtcdS3Timeout},
}
}
} else {
logrus.Info("ETCD snapshots are disabled")
}
......
......@@ -8,13 +8,13 @@ import (
"sort"
"strings"
"sync"
"time"
"github.com/k3s-io/k3s/pkg/generated/controllers/k3s.cattle.io"
"github.com/k3s-io/kine/pkg/endpoint"
"github.com/rancher/wharfie/pkg/registries"
"github.com/rancher/wrangler/v3/pkg/generated/controllers/core"
"github.com/rancher/wrangler/v3/pkg/leader"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
utilnet "k8s.io/apimachinery/pkg/util/net"
"k8s.io/apiserver/pkg/authentication/authenticator"
"k8s.io/client-go/tools/record"
......@@ -62,6 +62,21 @@ type Node struct {
DefaultRuntime string
}
type EtcdS3 struct {
AccessKey string `json:"accessKey,omitempty"`
Bucket string `json:"bucket,omitempty"`
ConfigSecret string `json:"configSecret,omitempty"`
Endpoint string `json:"endpoint,omitempty"`
EndpointCA string `json:"endpointCA,omitempty"`
Folder string `json:"folder,omitempty"`
Proxy string `json:"proxy,omitempty"`
Region string `json:"region,omitempty"`
SecretKey string `json:"secretKey,omitempty"`
Insecure bool `json:"insecure,omitempty"`
SkipSSLVerify bool `json:"skipSSLVerify,omitempty"`
Timeout metav1.Duration `json:"timeout,omitempty"`
}
type Containerd struct {
Address string
Log string
......@@ -216,27 +231,17 @@ type Control struct {
EncryptSkip bool
MinTLSVersion string
CipherSuites []string
TLSMinVersion uint16 `json:"-"`
TLSCipherSuites []uint16 `json:"-"`
EtcdSnapshotName string `json:"-"`
EtcdDisableSnapshots bool `json:"-"`
EtcdExposeMetrics bool `json:"-"`
EtcdSnapshotDir string `json:"-"`
EtcdSnapshotCron string `json:"-"`
EtcdSnapshotRetention int `json:"-"`
EtcdSnapshotCompress bool `json:"-"`
EtcdListFormat string `json:"-"`
EtcdS3 bool `json:"-"`
EtcdS3Endpoint string `json:"-"`
EtcdS3EndpointCA string `json:"-"`
EtcdS3SkipSSLVerify bool `json:"-"`
EtcdS3AccessKey string `json:"-"`
EtcdS3SecretKey string `json:"-"`
EtcdS3BucketName string `json:"-"`
EtcdS3Region string `json:"-"`
EtcdS3Folder string `json:"-"`
EtcdS3Timeout time.Duration `json:"-"`
EtcdS3Insecure bool `json:"-"`
TLSMinVersion uint16 `json:"-"`
TLSCipherSuites []uint16 `json:"-"`
EtcdSnapshotName string `json:"-"`
EtcdDisableSnapshots bool `json:"-"`
EtcdExposeMetrics bool `json:"-"`
EtcdSnapshotDir string `json:"-"`
EtcdSnapshotCron string `json:"-"`
EtcdSnapshotRetention int `json:"-"`
EtcdSnapshotCompress bool `json:"-"`
EtcdListFormat string `json:"-"`
EtcdS3 *EtcdS3 `json:"-"`
ServerNodeName string
VLevel int
VModule string
......
......@@ -12,7 +12,6 @@ import (
"net/url"
"os"
"path/filepath"
"regexp"
"sort"
"strconv"
"strings"
......@@ -26,6 +25,8 @@ import (
"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"
"github.com/k3s-io/k3s/pkg/etcd/s3"
"github.com/k3s-io/k3s/pkg/etcd/snapshot"
"github.com/k3s-io/k3s/pkg/server/auth"
"github.com/k3s-io/k3s/pkg/util"
"github.com/k3s-io/k3s/pkg/version"
......@@ -40,10 +41,8 @@ import (
"github.com/sirupsen/logrus"
"go.etcd.io/etcd/api/v3/etcdserverpb"
"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
"go.etcd.io/etcd/client/pkg/v3/logutil"
clientv3 "go.etcd.io/etcd/client/v3"
"go.etcd.io/etcd/etcdutl/v3/snapshot"
"go.uber.org/zap"
snapshotv3 "go.etcd.io/etcd/etcdutl/v3/snapshot"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
......@@ -93,8 +92,6 @@ var (
ErrAddressNotSet = errors.New("apiserver addresses not yet set")
ErrNotMember = errNotMember()
ErrMemberListFailed = errMemberListFailed()
invalidKeyChars = regexp.MustCompile(`[^-._a-zA-Z0-9]`)
)
type NodeControllerGetter func() controllerv1.NodeController
......@@ -110,8 +107,8 @@ type ETCD struct {
name string
address string
cron *cron.Cron
s3 *S3
cancel context.CancelFunc
s3 *s3.Controller
snapshotMu *sync.Mutex
}
......@@ -128,16 +125,16 @@ type Members struct {
Members []*etcdserverpb.Member `json:"members"`
}
type MembershipError struct {
Self string
Members []string
type membershipError struct {
self string
members []string
}
func (e *MembershipError) Error() string {
return fmt.Sprintf("this server is a not a member of the etcd cluster. Found %v, expect: %s", e.Members, e.Self)
func (e *membershipError) Error() string {
return fmt.Sprintf("this server is a not a member of the etcd cluster. Found %v, expect: %s", e.members, e.self)
}
func (e *MembershipError) Is(target error) bool {
func (e *membershipError) Is(target error) bool {
switch target {
case ErrNotMember:
return true
......@@ -145,17 +142,17 @@ func (e *MembershipError) Is(target error) bool {
return false
}
func errNotMember() error { return &MembershipError{} }
func errNotMember() error { return &membershipError{} }
type MemberListError struct {
Err error
type memberListError struct {
err error
}
func (e *MemberListError) Error() string {
return fmt.Sprintf("failed to get MemberList from server: %v", e.Err)
func (e *memberListError) Error() string {
return fmt.Sprintf("failed to get MemberList from server: %v", e.err)
}
func (e *MemberListError) Is(target error) bool {
func (e *memberListError) Is(target error) bool {
switch target {
case ErrMemberListFailed:
return true
......@@ -163,7 +160,7 @@ func (e *MemberListError) Is(target error) bool {
return false
}
func errMemberListFailed() error { return &MemberListError{} }
func errMemberListFailed() error { return &memberListError{} }
// NewETCD creates a new value of type
// ETCD with initialized cron and snapshot mutex values.
......@@ -256,7 +253,7 @@ func (e *ETCD) Test(ctx context.Context) error {
memberNameUrls = append(memberNameUrls, member.Name+"="+member.PeerURLs[0])
}
}
return &MembershipError{Members: memberNameUrls, Self: e.name + "=" + e.peerURL()}
return &membershipError{members: memberNameUrls, self: e.name + "=" + e.peerURL()}
}
// dbDir returns the path to dataDir/db/etcd
......@@ -391,14 +388,25 @@ func (e *ETCD) Reset(ctx context.Context, rebootstrap func() error) error {
// If asked to restore from a snapshot, do so
if e.config.ClusterResetRestorePath != "" {
if e.config.EtcdS3 {
if e.config.EtcdS3 != nil {
logrus.Infof("Retrieving etcd snapshot %s from S3", e.config.ClusterResetRestorePath)
if err := e.initS3IfNil(ctx); err != nil {
return err
s3client, err := e.getS3Client(ctx)
if err != nil {
if errors.Is(err, s3.ErrNoConfigSecret) {
return errors.New("cannot use S3 config secret when restoring snapshot; configuration must be set in CLI or config file")
} else {
return errors.Wrap(err, "failed to initialize S3 client")
}
}
if err := e.s3.Download(ctx); err != nil {
return err
dir, err := snapshotDir(e.config, true)
if err != nil {
return errors.Wrap(err, "failed to get the snapshot dir")
}
path, err := s3client.Download(ctx, e.config.ClusterResetRestorePath, dir)
if err != nil {
return errors.Wrap(err, "failed to download snapshot from S3")
}
e.config.ClusterResetRestorePath = path
logrus.Infof("S3 download complete for %s", e.config.ClusterResetRestorePath)
}
......@@ -442,6 +450,7 @@ func (e *ETCD) Start(ctx context.Context, clientAccessInfo *clientaccess.Info) e
}
go e.manageLearners(ctx)
go e.getS3Client(ctx)
if isInitialized {
// check etcd dir permission
......@@ -1416,7 +1425,7 @@ func ClientURLs(ctx context.Context, clientAccessInfo *clientaccess.Info, selfIP
// get the full list from the server we're joining
resp, err := clientAccessInfo.Get("/db/info")
if err != nil {
return nil, memberList, &MemberListError{Err: err}
return nil, memberList, &memberListError{err: err}
}
if err := json.Unmarshal(resp, &memberList); err != nil {
return nil, memberList, err
......@@ -1463,13 +1472,13 @@ func (e *ETCD) Restore(ctx context.Context) error {
}
var restorePath string
if strings.HasSuffix(e.config.ClusterResetRestorePath, compressedExtension) {
snapshotDir, err := snapshotDir(e.config, true)
if strings.HasSuffix(e.config.ClusterResetRestorePath, snapshot.CompressedExtension) {
dir, err := snapshotDir(e.config, true)
if err != nil {
return errors.Wrap(err, "failed to get the snapshot dir")
}
decompressSnapshot, err := e.decompressSnapshot(snapshotDir, e.config.ClusterResetRestorePath)
decompressSnapshot, err := e.decompressSnapshot(dir, e.config.ClusterResetRestorePath)
if err != nil {
return err
}
......@@ -1485,13 +1494,7 @@ func (e *ETCD) Restore(ctx context.Context) error {
}
logrus.Infof("Pre-restore etcd database moved to %s", oldDataDir)
lg, err := logutil.CreateDefaultZapLogger(zap.InfoLevel)
if err != nil {
return err
}
return snapshot.NewV3(lg).Restore(snapshot.RestoreConfig{
return snapshotv3.NewV3(e.client.GetLogger()).Restore(snapshotv3.RestoreConfig{
SnapshotPath: restorePath,
Name: e.name,
OutputDataDir: dbDir(e.config),
......
......@@ -11,8 +11,10 @@ import (
"github.com/k3s-io/k3s/pkg/clientaccess"
"github.com/k3s-io/k3s/pkg/daemons/config"
"github.com/k3s-io/k3s/pkg/etcd/s3"
testutil "github.com/k3s-io/k3s/tests"
"github.com/robfig/cron/v3"
"github.com/sirupsen/logrus"
clientv3 "go.etcd.io/etcd/client/v3"
"go.etcd.io/etcd/server/v3/etcdserver"
utilnet "k8s.io/apimachinery/pkg/util/net"
......@@ -47,10 +49,12 @@ func generateTestConfig() *config.Control {
EtcdSnapshotName: "etcd-snapshot",
EtcdSnapshotCron: "0 */12 * * *",
EtcdSnapshotRetention: 5,
EtcdS3Endpoint: "s3.amazonaws.com",
EtcdS3Region: "us-east-1",
SANs: []string{"127.0.0.1", mustGetAddress()},
CriticalControlArgs: criticalControlArgs,
EtcdS3: &config.EtcdS3{
Endpoint: "s3.amazonaws.com",
Region: "us-east-1",
},
SANs: []string{"127.0.0.1", mustGetAddress()},
CriticalControlArgs: criticalControlArgs,
}
}
......@@ -112,6 +116,10 @@ func Test_UnitETCD_IsInitialized(t *testing.T) {
want: false,
},
}
// enable logging
logrus.SetLevel(logrus.DebugLevel)
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
e := NewETCD()
......@@ -227,7 +235,7 @@ func Test_UnitETCD_Start(t *testing.T) {
name string
address string
cron *cron.Cron
s3 *S3
s3 *s3.Controller
}
type args struct {
clientAccessInfo *clientaccess.Info
......
package s3
import (
"encoding/base64"
"fmt"
"strconv"
"strings"
"time"
"github.com/k3s-io/k3s/pkg/daemons/config"
"github.com/k3s-io/k3s/pkg/util"
"github.com/sirupsen/logrus"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
var ErrNoConfigSecret = errNoConfigSecret()
type secretError struct {
err error
}
func (e *secretError) Error() string {
return fmt.Sprintf("failed to get etcd S3 config secret: %v", e.err)
}
func (e *secretError) Is(target error) bool {
switch target {
case ErrNoConfigSecret:
return true
}
return false
}
func errNoConfigSecret() error { return &secretError{} }
func (c *Controller) getConfigFromSecret(secretName string) (*config.EtcdS3, error) {
if c.core == nil {
return nil, &secretError{err: util.ErrCoreNotReady}
}
secret, err := c.core.V1().Secret().Get(metav1.NamespaceSystem, secretName, metav1.GetOptions{})
if err != nil {
return nil, &secretError{err: err}
}
etcdS3 := &config.EtcdS3{
AccessKey: string(secret.Data["etcd-s3-access-key"]),
Bucket: string(secret.Data["etcd-s3-bucket"]),
Endpoint: defaultEtcdS3.Endpoint,
Folder: string(secret.Data["etcd-s3-folder"]),
Proxy: string(secret.Data["etcd-s3-proxy"]),
Region: defaultEtcdS3.Region,
SecretKey: string(secret.Data["etcd-s3-secret-key"]),
Timeout: *defaultEtcdS3.Timeout.DeepCopy(),
}
// Set endpoint from secret if set
if v, ok := secret.Data["etcd-s3-endpoint"]; ok {
etcdS3.Endpoint = string(v)
}
// Set region from secret if set
if v, ok := secret.Data["etcd-s3-region"]; ok {
etcdS3.Region = string(v)
}
// Set timeout from secret if set
if v, ok := secret.Data["etcd-s3-timeout"]; ok {
if duration, err := time.ParseDuration(string(v)); err != nil {
logrus.Warnf("Failed to parse etcd-s3-timeout value from S3 config secret %s: %v", secretName, err)
} else {
etcdS3.Timeout.Duration = duration
}
}
// configure ssl verification, if value can be parsed
if v, ok := secret.Data["etcd-s3-skip-ssl-verify"]; ok {
if b, err := strconv.ParseBool(string(v)); err != nil {
logrus.Warnf("Failed to parse etcd-s3-skip-ssl-verify value from S3 config secret %s: %v", secretName, err)
} else {
etcdS3.SkipSSLVerify = b
}
}
// configure insecure http, if value can be parsed
if v, ok := secret.Data["etcd-s3-insecure"]; ok {
if b, err := strconv.ParseBool(string(v)); err != nil {
logrus.Warnf("Failed to parse etcd-s3-insecure value from S3 config secret %s: %v", secretName, err)
} else {
etcdS3.Insecure = b
}
}
// encode CA bundles from value, and keys in configmap if one is named
caBundles := []string{}
// Add inline CA bundle if set
if len(secret.Data["etcd-s3-endpoint-ca"]) > 0 {
caBundles = append(caBundles, base64.StdEncoding.EncodeToString(secret.Data["etcd-s3-endpoint-ca"]))
}
// Add CA bundles from named configmap if set
if caConfigMapName := string(secret.Data["etcd-s3-endpoint-ca-name"]); caConfigMapName != "" {
configMap, err := c.core.V1().ConfigMap().Get(metav1.NamespaceSystem, caConfigMapName, metav1.GetOptions{})
if err != nil {
logrus.Warnf("Failed to get ConfigMap %s for etcd-s3-endpoint-ca-name value from S3 config secret %s: %v", caConfigMapName, secretName, err)
} else {
for _, v := range configMap.Data {
caBundles = append(caBundles, base64.StdEncoding.EncodeToString([]byte(v)))
}
for _, v := range configMap.BinaryData {
caBundles = append(caBundles, base64.StdEncoding.EncodeToString(v))
}
}
}
// Concatenate all requested CA bundle strings into config var
etcdS3.EndpointCA = strings.Join(caBundles, " ")
return etcdS3, nil
}
package snapshot
import (
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"encoding/json"
"fmt"
"net/http"
"os"
"regexp"
"strings"
k3s "github.com/k3s-io/k3s/pkg/apis/k3s.cattle.io/v1"
"github.com/k3s-io/k3s/pkg/daemons/config"
"github.com/k3s-io/k3s/pkg/version"
"github.com/minio/minio-go/v7"
"github.com/sirupsen/logrus"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/util/validation"
"k8s.io/utils/ptr"
)
type SnapshotStatus string
const (
SuccessfulStatus SnapshotStatus = "successful"
FailedStatus SnapshotStatus = "failed"
CompressedExtension = ".zip"
MetadataDir = ".metadata"
)
var (
InvalidKeyChars = regexp.MustCompile(`[^-._a-zA-Z0-9]`)
LabelStorageNode = "etcd." + version.Program + ".cattle.io/snapshot-storage-node"
AnnotationTokenHash = "etcd." + version.Program + ".cattle.io/snapshot-token-hash"
ExtraMetadataConfigMapName = version.Program + "-etcd-snapshot-extra-metadata"
)
type S3Config struct {
config.EtcdS3
// Mask these fields in the embedded struct to avoid serializing their values in the snapshotFile record
AccessKey string `json:"accessKey,omitempty"`
ConfigSecret string `json:"configSecret,omitempty"`
Proxy string `json:"proxy,omitempty"`
SecretKey string `json:"secretKey,omitempty"`
Timeout metav1.Duration `json:"timeout,omitempty"`
}
// File represents a single snapshot and it's
// metadata.
type File struct {
Name string `json:"name"`
// Location contains the full path of the snapshot. For
// local paths, the location will be prefixed with "file://".
Location string `json:"location,omitempty"`
Metadata string `json:"metadata,omitempty"`
Message string `json:"message,omitempty"`
NodeName string `json:"nodeName,omitempty"`
CreatedAt *metav1.Time `json:"createdAt,omitempty"`
Size int64 `json:"size,omitempty"`
Status SnapshotStatus `json:"status,omitempty"`
S3 *S3Config `json:"s3Config,omitempty"`
Compressed bool `json:"compressed"`
// these fields are used for the internal representation of the snapshot
// to populate other fields before serialization to the legacy configmap.
MetadataSource *v1.ConfigMap `json:"-"`
NodeSource string `json:"-"`
TokenHash string `json:"-"`
}
// GenerateConfigMapKey generates a derived name for the snapshot that is safe for use
// as a configmap key.
func (sf *File) GenerateConfigMapKey() string {
name := InvalidKeyChars.ReplaceAllString(sf.Name, "_")
if sf.NodeName == "s3" {
return "s3-" + name
}
return "local-" + name
}
// GenerateName generates a derived name for the snapshot that is safe for use
// as a resource name.
func (sf *File) GenerateName() string {
name := strings.ToLower(sf.Name)
nodename := sf.NodeSource
if nodename == "" {
nodename = sf.NodeName
}
// Include a digest of the hostname and location to ensure unique resource
// names. Snapshots should already include the hostname, but this ensures we
// don't accidentally hide records if a snapshot with the same name somehow
// exists on multiple nodes.
digest := sha256.Sum256([]byte(nodename + sf.Location))
// If the lowercase filename isn't usable as a resource name, and short enough that we can include a prefix and suffix,
// generate a safe name derived from the hostname and timestamp.
if errs := validation.IsDNS1123Subdomain(name); len(errs) != 0 || len(name)+13 > validation.DNS1123SubdomainMaxLength {
nodename, _, _ := strings.Cut(nodename, ".")
name = fmt.Sprintf("etcd-snapshot-%s-%d", nodename, sf.CreatedAt.Unix())
if sf.Compressed {
name += CompressedExtension
}
}
if sf.NodeName == "s3" {
return "s3-" + name + "-" + hex.EncodeToString(digest[0:])[0:6]
}
return "local-" + name + "-" + hex.EncodeToString(digest[0:])[0:6]
}
// FromETCDSnapshotFile translates fields to the File from the ETCDSnapshotFile
func (sf *File) FromETCDSnapshotFile(esf *k3s.ETCDSnapshotFile) {
if esf == nil {
panic("cannot convert from nil ETCDSnapshotFile")
}
sf.Name = esf.Spec.SnapshotName
sf.Location = esf.Spec.Location
sf.CreatedAt = esf.Status.CreationTime
sf.NodeSource = esf.Spec.NodeName
sf.Compressed = strings.HasSuffix(esf.Spec.SnapshotName, CompressedExtension)
if esf.Status.ReadyToUse != nil && *esf.Status.ReadyToUse {
sf.Status = SuccessfulStatus
} else {
sf.Status = FailedStatus
}
if esf.Status.Size != nil {
sf.Size = esf.Status.Size.Value()
}
if esf.Status.Error != nil {
if esf.Status.Error.Time != nil {
sf.CreatedAt = esf.Status.Error.Time
}
message := "etcd snapshot failed"
if esf.Status.Error.Message != nil {
message = *esf.Status.Error.Message
}
sf.Message = base64.StdEncoding.EncodeToString([]byte(message))
}
if len(esf.Spec.Metadata) > 0 {
if b, err := json.Marshal(esf.Spec.Metadata); err != nil {
logrus.Warnf("Failed to marshal metadata for %s: %v", esf.Name, err)
} else {
sf.Metadata = base64.StdEncoding.EncodeToString(b)
}
}
if tokenHash := esf.Annotations[AnnotationTokenHash]; tokenHash != "" {
sf.TokenHash = tokenHash
}
if esf.Spec.S3 == nil {
sf.NodeName = esf.Spec.NodeName
} else {
sf.NodeName = "s3"
sf.S3 = &S3Config{
EtcdS3: config.EtcdS3{
Endpoint: esf.Spec.S3.Endpoint,
EndpointCA: esf.Spec.S3.EndpointCA,
SkipSSLVerify: esf.Spec.S3.SkipSSLVerify,
Bucket: esf.Spec.S3.Bucket,
Region: esf.Spec.S3.Region,
Folder: esf.Spec.S3.Prefix,
Insecure: esf.Spec.S3.Insecure,
},
}
}
}
// ToETCDSnapshotFile translates fields from the File to the ETCDSnapshotFile
func (sf *File) ToETCDSnapshotFile(esf *k3s.ETCDSnapshotFile) {
if esf == nil {
panic("cannot convert to nil ETCDSnapshotFile")
}
esf.Spec.SnapshotName = sf.Name
esf.Spec.Location = sf.Location
esf.Status.CreationTime = sf.CreatedAt
esf.Status.ReadyToUse = ptr.To(sf.Status == SuccessfulStatus)
esf.Status.Size = resource.NewQuantity(sf.Size, resource.DecimalSI)
if sf.NodeSource != "" {
esf.Spec.NodeName = sf.NodeSource
} else {
esf.Spec.NodeName = sf.NodeName
}
if sf.Message != "" {
var message string
b, err := base64.StdEncoding.DecodeString(sf.Message)
if err != nil {
logrus.Warnf("Failed to decode error message for %s: %v", sf.Name, err)
message = "etcd snapshot failed"
} else {
message = string(b)
}
esf.Status.Error = &k3s.ETCDSnapshotError{
Time: sf.CreatedAt,
Message: &message,
}
}
if sf.MetadataSource != nil {
esf.Spec.Metadata = sf.MetadataSource.Data
} else if sf.Metadata != "" {
metadata, err := base64.StdEncoding.DecodeString(sf.Metadata)
if err != nil {
logrus.Warnf("Failed to decode metadata for %s: %v", sf.Name, err)
} else {
if err := json.Unmarshal(metadata, &esf.Spec.Metadata); err != nil {
logrus.Warnf("Failed to unmarshal metadata for %s: %v", sf.Name, err)
}
}
}
if esf.ObjectMeta.Labels == nil {
esf.ObjectMeta.Labels = map[string]string{}
}
if esf.ObjectMeta.Annotations == nil {
esf.ObjectMeta.Annotations = map[string]string{}
}
if sf.TokenHash != "" {
esf.ObjectMeta.Annotations[AnnotationTokenHash] = sf.TokenHash
}
if sf.S3 == nil {
esf.ObjectMeta.Labels[LabelStorageNode] = esf.Spec.NodeName
} else {
esf.ObjectMeta.Labels[LabelStorageNode] = "s3"
esf.Spec.S3 = &k3s.ETCDSnapshotS3{
Endpoint: sf.S3.Endpoint,
EndpointCA: sf.S3.EndpointCA,
SkipSSLVerify: sf.S3.SkipSSLVerify,
Bucket: sf.S3.Bucket,
Region: sf.S3.Region,
Prefix: sf.S3.Folder,
Insecure: sf.S3.Insecure,
}
}
}
// Marshal returns the JSON encoding of the snapshot File, with metadata inlined as base64.
func (sf *File) Marshal() ([]byte, error) {
if sf.MetadataSource != nil {
if m, err := json.Marshal(sf.MetadataSource.Data); err != nil {
logrus.Debugf("Error attempting to marshal extra metadata contained in %s ConfigMap, error: %v", ExtraMetadataConfigMapName, err)
} else {
sf.Metadata = base64.StdEncoding.EncodeToString(m)
}
}
return json.Marshal(sf)
}
// IsNotExist returns true if the error is from http.StatusNotFound or os.IsNotExist
func IsNotExist(err error) bool {
if resp := minio.ToErrorResponse(err); resp.StatusCode == http.StatusNotFound || os.IsNotExist(err) {
return true
}
return false
}
......@@ -9,6 +9,7 @@ import (
"time"
apisv1 "github.com/k3s-io/k3s/pkg/apis/k3s.cattle.io/v1"
"github.com/k3s-io/k3s/pkg/etcd/snapshot"
controllersv1 "github.com/k3s-io/k3s/pkg/generated/controllers/k3s.cattle.io/v1"
"github.com/k3s-io/k3s/pkg/util"
"github.com/k3s-io/k3s/pkg/version"
......@@ -81,10 +82,10 @@ func (e *etcdSnapshotHandler) sync(key string, esf *apisv1.ETCDSnapshotFile) (*a
return nil, nil
}
sf := snapshotFile{}
sf.fromETCDSnapshotFile(esf)
sfKey := generateSnapshotConfigMapKey(sf)
m, err := marshalSnapshotFile(sf)
sf := &snapshot.File{}
sf.FromETCDSnapshotFile(esf)
sfKey := sf.GenerateConfigMapKey()
m, err := sf.Marshal()
if err != nil {
return nil, errors.Wrap(err, "failed to marshal snapshot ConfigMap data")
}
......@@ -283,9 +284,9 @@ func (e *etcdSnapshotHandler) reconcile() error {
// Ensure keys for existing snapshots
for sfKey, esf := range snapshots {
sf := snapshotFile{}
sf.fromETCDSnapshotFile(esf)
m, err := marshalSnapshotFile(sf)
sf := &snapshot.File{}
sf.FromETCDSnapshotFile(esf)
m, err := sf.Marshal()
if err != nil {
logrus.Warnf("Failed to marshal snapshot ConfigMap data for %s", sfKey)
continue
......@@ -327,12 +328,12 @@ func pruneConfigMap(snapshotConfigMap *v1.ConfigMap, pruneCount int) error {
return errors.New("unable to reduce snapshot ConfigMap size by eliding old snapshots")
}
var snapshotFiles []snapshotFile
var snapshotFiles []snapshot.File
retention := len(snapshotConfigMap.Data) - pruneCount
for name := range snapshotConfigMap.Data {
basename, compressed := strings.CutSuffix(name, compressedExtension)
basename, compressed := strings.CutSuffix(name, snapshot.CompressedExtension)
ts, _ := strconv.ParseInt(basename[strings.LastIndexByte(basename, '-')+1:], 10, 64)
snapshotFiles = append(snapshotFiles, snapshotFile{Name: name, CreatedAt: &metav1.Time{Time: time.Unix(ts, 0)}, Compressed: compressed})
snapshotFiles = append(snapshotFiles, snapshot.File{Name: name, CreatedAt: &metav1.Time{Time: time.Unix(ts, 0)}, Compressed: compressed})
}
// sort newest-first so we can prune entries past the retention count
......
......@@ -11,8 +11,8 @@ import (
"github.com/k3s-io/k3s/pkg/cluster/managed"
"github.com/k3s-io/k3s/pkg/daemons/config"
"github.com/k3s-io/k3s/pkg/util"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
type SnapshotOperation string
......@@ -24,21 +24,13 @@ const (
SnapshotOperationDelete SnapshotOperation = "delete"
)
type SnapshotRequestS3 struct {
s3Config
Timeout metav1.Duration `json:"timeout"`
AccessKey string `json:"accessKey"`
SecretKey string `json:"secretKey"`
}
type SnapshotRequest struct {
Operation SnapshotOperation `json:"operation"`
Name []string `json:"name,omitempty"`
Dir *string `json:"dir,omitempty"`
Compress *bool `json:"compress,omitempty"`
Retention *int `json:"retention,omitempty"`
S3 *SnapshotRequestS3 `json:"s3,omitempty"`
S3 *config.EtcdS3 `json:"s3,omitempty"`
ctx context.Context
}
......@@ -76,9 +68,12 @@ func (e *ETCD) snapshotHandler() http.Handler {
}
func (e *ETCD) handleList(rw http.ResponseWriter, req *http.Request) error {
if err := e.initS3IfNil(req.Context()); err != nil {
util.SendError(err, rw, req, http.StatusBadRequest)
return nil
if e.config.EtcdS3 != nil {
if _, err := e.getS3Client(req.Context()); err != nil {
err = errors.Wrap(err, "failed to initialize S3 client")
util.SendError(err, rw, req, http.StatusBadRequest)
return nil
}
}
sf, err := e.ListSnapshots(req.Context())
if sf == nil {
......@@ -90,9 +85,12 @@ func (e *ETCD) handleList(rw http.ResponseWriter, req *http.Request) error {
}
func (e *ETCD) handleSave(rw http.ResponseWriter, req *http.Request) error {
if err := e.initS3IfNil(req.Context()); err != nil {
util.SendError(err, rw, req, http.StatusBadRequest)
return nil
if e.config.EtcdS3 != nil {
if _, err := e.getS3Client(req.Context()); err != nil {
err = errors.Wrap(err, "failed to initialize S3 client")
util.SendError(err, rw, req, http.StatusBadRequest)
return nil
}
}
sr, err := e.Snapshot(req.Context())
if sr == nil {
......@@ -104,9 +102,12 @@ func (e *ETCD) handleSave(rw http.ResponseWriter, req *http.Request) error {
}
func (e *ETCD) handlePrune(rw http.ResponseWriter, req *http.Request) error {
if err := e.initS3IfNil(req.Context()); err != nil {
util.SendError(err, rw, req, http.StatusBadRequest)
return nil
if e.config.EtcdS3 != nil {
if _, err := e.getS3Client(req.Context()); err != nil {
err = errors.Wrap(err, "failed to initialize S3 client")
util.SendError(err, rw, req, http.StatusBadRequest)
return nil
}
}
sr, err := e.PruneSnapshots(req.Context())
if sr == nil {
......@@ -118,9 +119,12 @@ func (e *ETCD) handlePrune(rw http.ResponseWriter, req *http.Request) error {
}
func (e *ETCD) handleDelete(rw http.ResponseWriter, req *http.Request, snapshots []string) error {
if err := e.initS3IfNil(req.Context()); err != nil {
util.SendError(err, rw, req, http.StatusBadRequest)
return nil
if e.config.EtcdS3 != nil {
if _, err := e.getS3Client(req.Context()); err != nil {
err = errors.Wrap(err, "failed to initialize S3 client")
util.SendError(err, rw, req, http.StatusBadRequest)
return nil
}
}
sr, err := e.DeleteSnapshots(req.Context(), snapshots)
if sr == nil {
......@@ -149,7 +153,9 @@ func (e *ETCD) withRequest(sr *SnapshotRequest) *ETCD {
EtcdSnapshotCompress: e.config.EtcdSnapshotCompress,
EtcdSnapshotName: e.config.EtcdSnapshotName,
EtcdSnapshotRetention: e.config.EtcdSnapshotRetention,
EtcdS3: sr.S3,
},
s3: e.s3,
name: e.name,
address: e.address,
cron: e.cron,
......@@ -168,19 +174,6 @@ func (e *ETCD) withRequest(sr *SnapshotRequest) *ETCD {
if sr.Retention != nil {
re.config.EtcdSnapshotRetention = *sr.Retention
}
if sr.S3 != nil {
re.config.EtcdS3 = true
re.config.EtcdS3AccessKey = sr.S3.AccessKey
re.config.EtcdS3BucketName = sr.S3.Bucket
re.config.EtcdS3Endpoint = sr.S3.Endpoint
re.config.EtcdS3EndpointCA = sr.S3.EndpointCA
re.config.EtcdS3Folder = sr.S3.Folder
re.config.EtcdS3Insecure = sr.S3.Insecure
re.config.EtcdS3Region = sr.S3.Region
re.config.EtcdS3SecretKey = sr.S3.SecretKey
re.config.EtcdS3SkipSSLVerify = sr.S3.SkipSSLVerify
re.config.EtcdS3Timeout = sr.S3.Timeout.Duration
}
return re
}
......
......@@ -46,13 +46,8 @@ def provision(vm, role, role_num, node_num)
cluster-init: true
etcd-snapshot-schedule-cron: '*/1 * * * *'
etcd-snapshot-retention: 2
etcd-s3-insecure: true
etcd-s3-bucket: test-bucket
etcd-s3-folder: test-folder
etcd-s3: true
etcd-s3-endpoint: localhost:9090
etcd-s3-skip-ssl-verify: true
etcd-s3-access-key: test
etcd-s3-config-secret: k3s-etcd-s3-config
YAML
k3s.env = %W[K3S_KUBECONFIG_MODE=0644 #{install_type}]
k3s.config_mode = '0644' # side-step https://github.com/k3s-io/k3s/issues/4321
......
......@@ -87,7 +87,31 @@ var _ = Describe("Verify Create", Ordered, func() {
fmt.Println(res)
Expect(err).NotTo(HaveOccurred())
})
It("save s3 snapshot", func() {
It("save s3 snapshot using CLI", func() {
res, err := e2e.RunCmdOnNode("k3s etcd-snapshot save "+
"--etcd-s3-insecure=true "+
"--etcd-s3-bucket=test-bucket "+
"--etcd-s3-folder=test-folder "+
"--etcd-s3-endpoint=localhost:9090 "+
"--etcd-s3-skip-ssl-verify=true "+
"--etcd-s3-access-key=test ",
serverNodeNames[0])
Expect(err).NotTo(HaveOccurred())
Expect(res).To(ContainSubstring("Snapshot on-demand-server-0"))
})
It("creates s3 config secret", func() {
res, err := e2e.RunCmdOnNode("k3s kubectl create secret generic k3s-etcd-s3-config --namespace=kube-system "+
"--from-literal=etcd-s3-insecure=true "+
"--from-literal=etcd-s3-bucket=test-bucket "+
"--from-literal=etcd-s3-folder=test-folder "+
"--from-literal=etcd-s3-endpoint=localhost:9090 "+
"--from-literal=etcd-s3-skip-ssl-verify=true "+
"--from-literal=etcd-s3-access-key=test ",
serverNodeNames[0])
Expect(err).NotTo(HaveOccurred())
Expect(res).To(ContainSubstring("secret/k3s-etcd-s3-config created"))
})
It("save s3 snapshot using secret", func() {
res, err := e2e.RunCmdOnNode("k3s etcd-snapshot save", serverNodeNames[0])
Expect(err).NotTo(HaveOccurred())
Expect(res).To(ContainSubstring("Snapshot on-demand-server-0"))
......
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